allthingssecurity commented on code in PR #28319: URL: https://github.com/apache/camel/pull/28319#discussion_r4358765727
########## components/camel-tahu/src/test/java/org/apache/camel/component/tahu/FakeMqttBroker.java: ########## @@ -0,0 +1,168 @@ +/* + * 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.tahu; + +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * A minimal in-process MQTT 3.1.1 broker for tests: it accepts every connection and subscription, acknowledges every + * publish and records the topics published to it. It does not route messages to subscribers. + */ +final class FakeMqttBroker implements AutoCloseable { + + private final ServerSocket serverSocket; + private final List<Socket> sockets = new CopyOnWriteArrayList<>(); + private final List<String> publishedTopics = new CopyOnWriteArrayList<>(); + private final AtomicInteger connects = new AtomicInteger(); + private volatile boolean running = true; + + FakeMqttBroker() throws IOException { + serverSocket = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + Thread acceptor = new Thread(this::accept, "fake-mqtt-acceptor"); + acceptor.setDaemon(true); + acceptor.start(); + } + + int getPort() { + return serverSocket.getLocalPort(); + } + + List<String> getPublishedTopics() { + return publishedTopics; + } + + int getConnects() { + return connects.get(); + } + + private void accept() { + while (running) { + try { + Socket socket = serverSocket.accept(); + sockets.add(socket); + Thread handler = new Thread(() -> handle(socket), "fake-mqtt-connection"); + handler.setDaemon(true); + handler.start(); + } catch (IOException e) { + return; + } + } + } + + private void handle(Socket socket) { + try (socket) { + DataInputStream in = new DataInputStream(socket.getInputStream()); + OutputStream out = socket.getOutputStream(); + while (running) { + int header = in.read(); + if (header < 0) { + return; + } + byte[] body = new byte[readRemainingLength(in)]; + in.readFully(body); + int type = header >> 4; + switch (type) { + case 1 -> { + connects.incrementAndGet(); + write(out, 0x20, 0, 0); + } + case 3 -> { + int qos = (header >> 1) & 3; + int topicLength = ((body[0] & 0xff) << 8) | (body[1] & 0xff); + publishedTopics.add(new String(body, 2, topicLength, StandardCharsets.UTF_8)); + if (qos == 1) { + write(out, 0x40, body[2 + topicLength], body[3 + topicLength]); + } else if (qos == 2) { + write(out, 0x50, body[2 + topicLength], body[3 + topicLength]); + } + } + case 6 -> write(out, 0x70, body[0], body[1]); + case 8 -> { + // SUBACK: the packet id and the requested QoS of every topic filter + ByteArrayOutputStream granted = new ByteArrayOutputStream(); + int i = 2; + while (i < body.length) { + int length = ((body[i] & 0xff) << 8) | (body[i + 1] & 0xff); + i += 2 + length; + granted.write(body[i++]); + } + byte[] qos = granted.toByteArray(); + byte[] suback = new byte[2 + qos.length]; + suback[0] = body[0]; + suback[1] = body[1]; + System.arraycopy(qos, 0, suback, 2, qos.length); + writePacket(out, 0x90, suback); + } + case 10 -> write(out, 0xB0, body[0], body[1]); + case 12 -> writePacket(out, 0xD0, new byte[0]); + case 14 -> { + return; + } + default -> { + // nothing to answer + } + } + } + } catch (IOException e) { + // the client went away + } finally { + sockets.remove(socket); + } + } + + private static int readRemainingLength(DataInputStream in) throws IOException { + int value = 0; + int multiplier = 1; + int digit; + do { + digit = in.readUnsignedByte(); + value += (digit & 127) * multiplier; + multiplier *= 128; + } while ((digit & 128) != 0); + return value; + } + + private static void write(OutputStream out, int header, int b1, int b2) throws IOException { + writePacket(out, header, new byte[] { (byte) b1, (byte) b2 }); + } + + private static synchronized void writePacket(OutputStream out, int header, byte[] body) throws IOException { + out.write(header); + out.write(body.length); Review Comment: Left as is, as @davsclaus agreed: each test packet is far below 128 bytes, so the one-byte remaining length is correct for every packet the fake broker sends. _Claude Code on behalf of allthingssecurity_ ########## components/camel-tahu/src/test/java/org/apache/camel/component/tahu/FakeMqttBroker.java: ########## @@ -0,0 +1,168 @@ +/* + * 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.tahu; + +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * A minimal in-process MQTT 3.1.1 broker for tests: it accepts every connection and subscription, acknowledges every + * publish and records the topics published to it. It does not route messages to subscribers. + */ +final class FakeMqttBroker implements AutoCloseable { + + private final ServerSocket serverSocket; + private final List<Socket> sockets = new CopyOnWriteArrayList<>(); + private final List<String> publishedTopics = new CopyOnWriteArrayList<>(); + private final AtomicInteger connects = new AtomicInteger(); + private volatile boolean running = true; + + FakeMqttBroker() throws IOException { + serverSocket = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + Thread acceptor = new Thread(this::accept, "fake-mqtt-acceptor"); + acceptor.setDaemon(true); + acceptor.start(); + } + + int getPort() { + return serverSocket.getLocalPort(); + } + + List<String> getPublishedTopics() { + return publishedTopics; + } + + int getConnects() { + return connects.get(); + } + + private void accept() { + while (running) { + try { + Socket socket = serverSocket.accept(); + sockets.add(socket); + Thread handler = new Thread(() -> handle(socket), "fake-mqtt-connection"); + handler.setDaemon(true); + handler.start(); + } catch (IOException e) { + return; + } + } + } + + private void handle(Socket socket) { + try (socket) { + DataInputStream in = new DataInputStream(socket.getInputStream()); + OutputStream out = socket.getOutputStream(); + while (running) { + int header = in.read(); + if (header < 0) { + return; + } + byte[] body = new byte[readRemainingLength(in)]; + in.readFully(body); + int type = header >> 4; + switch (type) { + case 1 -> { + connects.incrementAndGet(); + write(out, 0x20, 0, 0); + } + case 3 -> { + int qos = (header >> 1) & 3; + int topicLength = ((body[0] & 0xff) << 8) | (body[1] & 0xff); + publishedTopics.add(new String(body, 2, topicLength, StandardCharsets.UTF_8)); + if (qos == 1) { + write(out, 0x40, body[2 + topicLength], body[3 + topicLength]); + } else if (qos == 2) { + write(out, 0x50, body[2 + topicLength], body[3 + topicLength]); + } + } + case 6 -> write(out, 0x70, body[0], body[1]); + case 8 -> { + // SUBACK: the packet id and the requested QoS of every topic filter + ByteArrayOutputStream granted = new ByteArrayOutputStream(); + int i = 2; + while (i < body.length) { + int length = ((body[i] & 0xff) << 8) | (body[i + 1] & 0xff); + i += 2 + length; + granted.write(body[i++]); + } + byte[] qos = granted.toByteArray(); + byte[] suback = new byte[2 + qos.length]; + suback[0] = body[0]; + suback[1] = body[1]; + System.arraycopy(qos, 0, suback, 2, qos.length); + writePacket(out, 0x90, suback); + } + case 10 -> write(out, 0xB0, body[0], body[1]); + case 12 -> writePacket(out, 0xD0, new byte[0]); + case 14 -> { + return; + } + default -> { + // nothing to answer + } + } + } + } catch (IOException e) { + // the client went away + } finally { + sockets.remove(socket); + } + } + + private static int readRemainingLength(DataInputStream in) throws IOException { + int value = 0; + int multiplier = 1; + int digit; + do { + digit = in.readUnsignedByte(); + value += (digit & 127) * multiplier; + multiplier *= 128; + } while ((digit & 128) != 0); + return value; + } + + private static void write(OutputStream out, int header, int b1, int b2) throws IOException { + writePacket(out, header, new byte[] { (byte) b1, (byte) b2 }); + } + + private static synchronized void writePacket(OutputStream out, int header, byte[] body) throws IOException { Review Comment: Left as is, as @davsclaus agreed: each socket has one writer, and each test starts its own broker, so the class-wide lock serialises nothing that runs in parallel. _Claude Code on behalf of allthingssecurity_ ########## components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuEdgeProducer.java: ########## @@ -142,6 +189,45 @@ private TahuEdgeClient createClient(TahuEdgeEndpoint endpoint, EdgeNodeDescripto return tahuEdgeClient; } + @Override + protected void doStart() throws Exception { + super.doStart(); + + SparkplugBPayloadMap metricMap = endpoint.getMetricDataTypePayloadMap(); + if (metricMap != null) { + synchronized (descriptorMetricMaps) { + if (descriptorMetricMaps.get(edgeNodeDescriptor) == metricMap) { + descriptorMetricMapUsers.merge(edgeNodeDescriptor, 1, Integer::sum); + } else { + descriptorMetricMaps.put(edgeNodeDescriptor, metricMap); + descriptorMetricMapUsers.put(edgeNodeDescriptor, 1); + } + } + // the client of the Edge Node may have been created again while this producer was stopped + TahuEdgeClient client = descriptorClients.get(clientDescriptor); + if (client != null) { + client.addDeviceMetricDataPayloadMap(edgeNodeDescriptor, metricMap); Review Comment: Agreed that the shut-down case is harmless. Claus found the other side of this: a Device that starts while the Edge Node runs `computeIfAbsent` in `createClient` gets null from `descriptorClients.get` and adds its map to no client, so the new client has no map for that Device. Both go into the drafted follow-up JIRA, fixed by the one lock around the start and stop of an Edge Node, which the Device producer also takes when it adds its map. _Claude Code on behalf of allthingssecurity_ -- 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]
