From edcc06ea7b539637fc1c0eea98962b36175020ac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Toma=C5=BE=20Su=C5=A1nik?= Date: Mon, 9 Nov 2026 13:03:10 +0100 Subject: [PATCH] Client: unsubscribe the topic of the removed subscription OpenCmwDataSource.unsubscribe looked up the subscription's endpoint, then used the endpoint the data source was created with. That is the endpoint of the first subscription on the connection. Every unsubscribe removed the first subscription's topic: the closed subscription kept receiving, logged as "Could not find subscription", and the first subscription went silent. --- .../src/main/java/io/opencmw/client/OpenCmwDataSource.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/client/src/main/java/io/opencmw/client/OpenCmwDataSource.java b/client/src/main/java/io/opencmw/client/OpenCmwDataSource.java index 592a4358..66ddc198 100644 --- a/client/src/main/java/io/opencmw/client/OpenCmwDataSource.java +++ b/client/src/main/java/io/opencmw/client/OpenCmwDataSource.java @@ -334,12 +334,12 @@ public void unsubscribe(final String reqId) { final URI subscriptionEndpoint = subscriptions.remove(reqId); final byte[] serviceId = subscriptionEndpoint.getPath().substring(1).getBytes(UTF_8); if (socket.getSocketType() == SocketType.DEALER) { // mdp - final var msg = new MdpMessage(clientId.getBytes(UTF_8), PROT_CLIENT, UNSUBSCRIBE, serviceId, reqId.getBytes(UTF_8), endpoint, EMPTY_FRAME, "", null); + final var msg = new MdpMessage(clientId.getBytes(UTF_8), PROT_CLIENT, UNSUBSCRIBE, serviceId, reqId.getBytes(UTF_8), subscriptionEndpoint, EMPTY_FRAME, "", null); if (!msg.send(socket)) { - LOGGER.atError().addArgument(reqId).addArgument(endpoint).log("unsubscribe error (reqId: {}) for endpoint: {}"); + LOGGER.atError().addArgument(reqId).addArgument(subscriptionEndpoint).log("unsubscribe error (reqId: {}) for endpoint: {}"); } } else { // mds - final String id = getZMQTopicFromURI(endpoint); + final String id = getZMQTopicFromURI(subscriptionEndpoint); socket.unsubscribe(id.getBytes(UTF_8)); } }