etzel-mes commented on code in PR #28631: URL: https://github.com/apache/camel/pull/28631#discussion_r4403212973
########## components/camel-mtconnect/src/main/java/org/apache/camel/component/mtconnect/AgentProtocol.java: ########## @@ -0,0 +1,453 @@ +/* + * 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.mtconnect; + +import java.time.Instant; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.camel.Exchange; +import org.apache.camel.support.DefaultConsumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * The MTConnect protocol against one agent, one step at a time: probe, current, with assets=true the assets, then + * sample, either the long-lived streaming request or one sample request per step. It resynchronises from current when + * the agent's buffer no longer holds the next observation, re-reads the probe when the agent restarts, and reports an + * outage once per data item. Each observation becomes one exchange of the consumer, processed synchronously in sequence + * order. The streaming and the polling consumer drive it; they own only the threading or the scheduling. Not + * thread-safe: one step runs at a time. + */ +final class AgentProtocol { + + private static final Logger LOG = LoggerFactory.getLogger(AgentProtocol.class); + + private enum Phase { + PROBE, + CURRENT, + ASSETS, + SAMPLE + } + + private static final String ASSET_CHANGED = "AssetChanged"; + private static final String ASSET_REMOVED = "AssetRemoved"; + private static final String ASSET_NOT_FOUND = "ASSET_NOT_FOUND"; + private static final String GONE = "gone"; + + private final MTConnectEndpoint endpoint; + private final DefaultConsumer consumer; + private final AgentReader reader; + private final boolean streaming; + + private Phase phase = Phase.PROBE; + private Map<String, DataItem> dataItems = Map.of(); + private final Set<String> known = new LinkedHashSet<>(); + private final Set<String> bad = new HashSet<>(); + private Long instanceId; + private long cursor; + private long count; + private String resync; + private boolean outage; + private int delivered; + // asset tracking: the asset buffer size from the probe, the device names by uuid, the AssetChanged and + // AssetRemoved observations whose asset is still to be read, and the version last sent per asset id + private Long assetBufferSize; + private Map<String, String> deviceNames = Map.of(); + private final Map<String, Observation> pendingAssets = new LinkedHashMap<>(); + private final Map<String, String> sentAssets = new HashMap<>(); + + /** + * @param streaming whether the sample phase reads the streaming request, or one sample request per step + */ + AgentProtocol(MTConnectEndpoint endpoint, DefaultConsumer consumer, AgentReader reader, boolean streaming) { + this.endpoint = endpoint; + this.consumer = consumer; + this.reader = reader; + this.streaming = streaming; + } + + /** + * Performs the next step: probe, current, assets or sample. A failure is reported here, as an outage or to the + * exception handler. + * + * @return true when the next step may follow at once; false after a failure, or when the agent ended the sample + * stream, so that the caller waits before the next step + */ + boolean step() throws InterruptedException { + delivered = 0; + try { + switch (phase) { + case PROBE -> probe(); + case CURRENT -> current(); + case ASSETS -> assets(); + case SAMPLE -> { + if (!readPendingAssets()) { + return true; + } + return streaming ? stream() : poll(); + } + } + return true; + } catch (AgentFailure failure) { + if (consumer.isRunAllowed()) { + onFailure(failure); + } + return false; + } catch (RuntimeException e) { + consumer.getExceptionHandler().handleException("Unexpected failure reading the MTConnect agent " + + endpoint.getAgentUrl(), + e); + return false; + } + } + + /** The number of messages the last step delivered. */ + int delivered() { + return delivered; + } + + private void probe() throws AgentFailure, InterruptedException { + AgentReader.Answer answer = reader.probe(); + if (answer.document() instanceof AgentDocument.Error error) { + throw agentError(error); + } + AgentDocument.Probe probe = answer.expect(AgentDocument.Probe.class); + Long probeInstance = probe.header().instanceId(); + if (instanceId != null && !instanceId.equals(probeInstance)) { + resync = MTConnectConstants.RESYNC_INSTANCE_CHANGED; + } + instanceId = probeInstance; + dataItems = probe.dataItems(); + deviceNames = new HashMap<>(); + dataItems.values().forEach(item -> deviceNames.put(item.deviceUuid(), item.deviceName())); + assetBufferSize = probe.header().assetBufferSize(); + pendingAssets.clear(); + known.clear(); Review Comment: Deliberate. `sentAssets` is the memory of what the route has already received, and a restart does not change an asset the agent still holds unchanged. After a re-probe the protocol reads the assets again (`anAgentRestartReadsTheAssetsAgain` counts the second `/assets` request) and sends only those whose timestamp or removed state changed, so the route does not get every cutting tool a second time whenever the agent restarts. An asset id the agent did not have before is a new key and is delivered. An asset whose content changes while its timestamp does not would be an agent defect the component cannot see on any path. The map holds one entry per asset id ever seen, bounded by the agent's asset buffer; clearing it on re-probe would trade that for a duplicate of the whole asset set on every restart. _Claude Code on behalf of etzel-mes_ ########## components/camel-mtconnect/src/main/java/org/apache/camel/component/mtconnect/MultipartReader.java: ########## @@ -0,0 +1,185 @@ +/* + * 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.mtconnect; + +import java.io.ByteArrayOutputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Locale; + +/** + * Reads the parts of an agent's streaming response, {@code multipart/mixed} (2.x agents) or + * {@code multipart/x-mixed-replace} (1.x agents). Part header names are compared without case, because 1.x agents write + * {@code Content-length}. A part with a Content-Length header is complete after that many bytes, so it is delivered + * before the next boundary arrives; a part without one ends at the next boundary. Agents end a stream without a closing + * boundary, so the end of the input between parts is the end of the stream. The reader reads byte by byte, so its input + * must be buffered. + */ +final class MultipartReader { + + private static final int MAX_LINE = 8192; + + private final InputStream in; + private final byte[] delimiter; + private final String boundaryLine; + private boolean atBoundary; + private boolean finished; + + MultipartReader(InputStream in, String boundary) { + this.in = in; + this.boundaryLine = "--" + boundary; + this.delimiter = ("\r\n--" + boundary).getBytes(StandardCharsets.US_ASCII); + } + + /** + * The boundary parameter of a multipart content type, or null when the content type is not multipart. + */ + static String boundary(String contentType) { + if (contentType == null) { + return null; + } + String[] parameters = contentType.split(";"); + if (!parameters[0].trim().toLowerCase(Locale.ROOT).startsWith("multipart/")) { + return null; + } + for (int i = 1; i < parameters.length; i++) { + String parameter = parameters[i].trim(); + int equals = parameter.indexOf('='); + if (equals > 0 && parameter.substring(0, equals).trim().equalsIgnoreCase("boundary")) { + String value = parameter.substring(equals + 1).trim(); + if (value.length() >= 2 && value.startsWith("\"") && value.endsWith("\"")) { + value = value.substring(1, value.length() - 1); + } + return value.isEmpty() ? null : value; + } + } + return null; + } + + /** + * The content of the next part, or null when the stream ended between parts. + * + * @throws IOException when the stream ends inside a part or the framing is invalid + */ + byte[] nextPart() throws IOException { + if (finished) { + return null; + } + if (!atBoundary && !skipToBoundary()) { + finished = true; + return null; + } + atBoundary = false; + long contentLength = -1; + while (true) { + String line = readLine(); + if (line == null) { + throw new EOFException("The stream ended inside the headers of a part"); + } + if (line.isEmpty()) { + break; + } + int colon = line.indexOf(':'); + if (colon > 0 && line.substring(0, colon).trim().equalsIgnoreCase("Content-Length")) { + try { + contentLength = Long.parseLong(line.substring(colon + 1).trim()); + } catch (NumberFormatException e) { + throw new IOException("Invalid part header: " + line, e); + } + } + } + return contentLength >= 0 ? readExactly(contentLength) : readToDelimiter(); + } + + /** Skips lines up to the next boundary line; false at the end of the stream or after the closing boundary. */ + private boolean skipToBoundary() throws IOException { + while (true) { + String line = readLine(); + if (line == null) { + return false; + } + String trimmed = line.stripTrailing(); + if (trimmed.equals(boundaryLine + "--")) { + return false; + } + if (trimmed.equals(boundaryLine)) { + return true; + } + } + } + + private byte[] readExactly(long length) throws IOException { + if (length > Integer.MAX_VALUE - 8) { + throw new IOException("Part too large: " + length + " bytes"); + } + byte[] content = in.readNBytes((int) length); Review Comment: `InputStream.readNBytes(int)` reads in chunks of 16 KiB and assembles the result at the end, so a spoofed `Content-Length` costs heap only for the bytes the agent actually sends, and the silence bound (`ReadDeadline`) ends a part that stops arriving. A part without `Content-Length`, and a whole document answer, grow the same way, so a cap on this one path would not bound the component's memory. The agent is on the machine network with the component's own trust in it; we have left the size to the agent's limits and would add a `maxPartSize` option if the committers want one. _Claude Code on behalf of etzel-mes_ -- 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]
