Add topic filter for event WebSocket (#4550)

* Add topic filter for event WebSocket

Signed-off-by: Florian Hotze <dev@florianhotze.com>

bundles/org.openhab.core.io.websocket/src/main/java/org/openhab/core/io/websocket/event/EventWebSocket.java

@@ -16,6 +16,7 @@ import java.io.IOException;
 import java.lang.reflect.Type;
 import java.util.List;
 import java.util.Objects;
+import java.util.regex.Pattern;
 
 import org.eclipse.jdt.annotation.NonNullByDefault;
 import org.eclipse.jdt.annotation.Nullable;
@@ -29,6 +30,7 @@ import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
 import org.eclipse.jetty.websocket.api.annotations.WebSocket;
 import org.openhab.core.events.Event;
 import org.openhab.core.events.EventPublisher;
+import org.openhab.core.events.TopicEventFilter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -40,6 +42,7 @@ import com.google.gson.reflect.TypeToken;
  * The {@link EventWebSocket} is the WebSocket implementation that extends the event bus
  *
  * @author Jan N. Klug - Initial contribution
+ * @author Florian Hotze - Add topic filter
  */
 @WebSocket
 @NonNullByDefault
@@ -49,6 +52,7 @@ public class EventWebSocket {
     public static final String WEBSOCKET_TOPIC_PREFIX = "openhab/websocket/";
 
     private static final Type STRING_LIST_TYPE = TypeToken.getParameterized(List.class, String.class).getType();
+    private static final Pattern TOPIC_VALIDATE_PATTERN = Pattern.compile("^!?(\\w*\\*?\\/?)+$");
 
     private final Logger logger = LoggerFactory.getLogger(EventWebSocket.class);
 
@@ -63,6 +67,8 @@ public class EventWebSocket {
 
     private List<String> typeFilter = List.of();
     private List<String> sourceFilter = List.of();
+    private @Nullable TopicEventFilter topicIncludeFilter = null;
+    private @Nullable TopicEventFilter topicExcludeFilter = null;
 
     public EventWebSocket(Gson gson, EventWebSocketAdapter wsAdapter, ItemEventUtility itemEventUtility,
             EventPublisher eventPublisher) {
@@ -148,6 +154,32 @@ public class EventWebSocket {
                                     remoteEndpoint.getInetSocketAddress(), typeFilter);
                             responseEvent = new EventDTO(WEBSOCKET_EVENT_TYPE, WEBSOCKET_TOPIC_PREFIX + "filter/source",
                                     eventDTO.payload, null, eventDTO.eventId);
+                        } else if ((WEBSOCKET_TOPIC_PREFIX + "filter/topic").equals(eventDTO.topic)) {
+                            List<String> topics = Objects
+                                    .requireNonNullElse(gson.fromJson(eventDTO.payload, STRING_LIST_TYPE), List.of());
+                            for (String topic : topics) {
+                                if (!TOPIC_VALIDATE_PATTERN.matcher(topic).matches()) {
+                                    throw new EventProcessingException(
+                                            "Invalid topic '" + topic + "' in topic filter WebSocketEvent");
+                                }
+                            }
+                            List<String> includeTopics = topics.stream().filter(t -> !t.startsWith("!")).toList();
+                            List<String> excludeTopics = topics.stream().filter(t -> t.startsWith("!"))
+                                    .map(t -> t.substring(1)).toList();
+                            // convert to regex: replace any wildcard (*) with the regex pattern (.*)
+                            includeTopics = includeTopics.stream().map(t -> t.trim().replace("*", ".*") + "$").toList();
+                            excludeTopics = excludeTopics.stream().map(t -> t.trim().replace("*", ".*") + "$").toList();
+                            // create topic filter if topic list not empty
+                            if (!includeTopics.isEmpty()) {
+                                topicIncludeFilter = new TopicEventFilter(includeTopics);
+                            }
+                            if (!excludeTopics.isEmpty()) {
+                                topicExcludeFilter = new TopicEventFilter(excludeTopics);
+                            }
+                            logger.debug("Setting topic filter for connection to {}: {}",
+                                    remoteEndpoint.getInetSocketAddress(), topics);
+                            responseEvent = new EventDTO(WEBSOCKET_EVENT_TYPE, WEBSOCKET_TOPIC_PREFIX + "filter/topic",
+                                    eventDTO.payload, null, eventDTO.eventId);
                         } else {
                             throw new EventProcessingException("Invalid topic or payload in WebSocketEvent");
                         }
@@ -195,7 +227,9 @@ public class EventWebSocket {
         try {
             String source = event.getSource();
             if ((source == null || !sourceFilter.contains(event.getSource()))
-                    && (typeFilter.isEmpty() || typeFilter.contains(event.getType()))) {
+                    && (typeFilter.isEmpty() || typeFilter.contains(event.getType()))
+                    && (topicIncludeFilter == null || topicIncludeFilter.apply(event))
+                    && (topicExcludeFilter == null || !topicExcludeFilter.apply(event))) {
                 sendMessage(gson.toJson(new EventDTO(event)));
             }
         } catch (IOException e) {

bundles/org.openhab.core.io.websocket/src/test/java/org/openhab/core/io/websocket/EventWebSocketTest.java

@@ -14,6 +14,7 @@ package org.openhab.core.io.websocket;
 
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.clearInvocations;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -55,6 +56,7 @@ import com.google.gson.Gson;
  * The {@link EventWebSocketTest} contains tests for the {@link EventWebSocket}
  *
  * @author Jan N. Klug - Initial contribution
+ * @author Florian Hotze - Add topic filter tests
  */
 @NonNullByDefault
 @ExtendWith(MockitoExtension.class)
@@ -240,6 +242,87 @@ public class EventWebSocketTest {
         verify(remoteEndpoint, times(2)).sendString(any());
     }
 
+    @Test
+    public void eventFromBusFilterIncludeTopic() throws IOException {
+        EventDTO eventDTO = new EventDTO(WEBSOCKET_EVENT_TYPE, WEBSOCKET_TOPIC_PREFIX + "filter/topic",
+                "[\"openhab/items/*/command\", \"openhab/items/*/statechanged\"]", null, null);
+        EventDTO responseEventDTO = new EventDTO(WEBSOCKET_EVENT_TYPE, WEBSOCKET_TOPIC_PREFIX + "filter/topic",
+                eventDTO.payload, null, null);
+        eventWebSocket.onText(gson.toJson(eventDTO));
+        verify(remoteEndpoint).sendString(gson.toJson(responseEventDTO));
+        clearInvocations(remoteEndpoint);
+
+        // subscribed topics are sent
+        Event event = ItemEventFactory.createCommandEvent(TEST_ITEM_NAME, DecimalType.ZERO,
+                REMOTE_WEBSOCKET_IMPLEMENTATION);
+        eventWebSocket.processEvent(event);
+        verify(remoteEndpoint).sendString(gson.toJson(new EventDTO(event)));
+
+        event = ItemEventFactory.createStateChangedEvent(TEST_ITEM_NAME, DecimalType.ZERO, DecimalType.ZERO);
+        eventWebSocket.processEvent(event);
+        verify(remoteEndpoint).sendString(gson.toJson(new EventDTO(event)));
+
+        // not subscribed topics are not sent
+        event = ItemEventFactory.createStateEvent(TEST_ITEM_NAME, DecimalType.ZERO, REMOTE_WEBSOCKET_IMPLEMENTATION);
+        eventWebSocket.processEvent(event);
+        verify(remoteEndpoint, times(2)).sendString(any());
+    }
+
+    @Test
+    public void eventFromBusFilterExcludeTopic() throws IOException {
+        EventDTO eventDTO = new EventDTO(WEBSOCKET_EVENT_TYPE, WEBSOCKET_TOPIC_PREFIX + "filter/topic",
+                "[\"!openhab/items/" + TEST_ITEM_NAME + "/command\"]", null, null);
+        EventDTO responseEventDTO = new EventDTO(WEBSOCKET_EVENT_TYPE, WEBSOCKET_TOPIC_PREFIX + "filter/topic",
+                eventDTO.payload, null, null);
+        eventWebSocket.onText(gson.toJson(eventDTO));
+        verify(remoteEndpoint).sendString(gson.toJson(responseEventDTO));
+        clearInvocations(remoteEndpoint);
+
+        // excluded topics are not sent
+        Event event = ItemEventFactory.createCommandEvent(TEST_ITEM_NAME, DecimalType.ZERO,
+                REMOTE_WEBSOCKET_IMPLEMENTATION);
+        eventWebSocket.processEvent(event);
+        verify(remoteEndpoint, times(0)).sendString(any());
+
+        // not excluded topics are sent
+        event = ItemEventFactory.createStateChangedEvent(TEST_ITEM_NAME, DecimalType.ZERO, DecimalType.ZERO);
+        eventWebSocket.processEvent(event);
+        verify(remoteEndpoint).sendString(gson.toJson(new EventDTO(event)));
+
+        event = ItemEventFactory.createStateEvent(TEST_ITEM_NAME, DecimalType.ZERO, REMOTE_WEBSOCKET_IMPLEMENTATION);
+        eventWebSocket.processEvent(event);
+        verify(remoteEndpoint).sendString(gson.toJson(new EventDTO(event)));
+
+        event = ItemEventFactory.createStateEvent("anotherItem", DecimalType.ZERO, REMOTE_WEBSOCKET_IMPLEMENTATION);
+        eventWebSocket.processEvent(event);
+        verify(remoteEndpoint).sendString(gson.toJson(new EventDTO(event)));
+    }
+
+    @Test
+    public void eventFromBusFilterIncludeAndExcludeTopic() throws IOException {
+        EventDTO eventDTO = new EventDTO(WEBSOCKET_EVENT_TYPE, WEBSOCKET_TOPIC_PREFIX + "filter/topic",
+                "[\"openhab/items/*/*\", \"!openhab/items/*/command\"]", null, null);
+        EventDTO responseEventDTO = new EventDTO(WEBSOCKET_EVENT_TYPE, WEBSOCKET_TOPIC_PREFIX + "filter/topic",
+                eventDTO.payload, null, null);
+        eventWebSocket.onText(gson.toJson(eventDTO));
+        verify(remoteEndpoint).sendString(gson.toJson(responseEventDTO));
+        clearInvocations(remoteEndpoint);
+
+        // included topics are sent
+        Event event = ItemEventFactory.createStateChangedEvent(TEST_ITEM_NAME, DecimalType.ZERO, DecimalType.ZERO);
+        eventWebSocket.processEvent(event);
+        verify(remoteEndpoint).sendString(gson.toJson(new EventDTO(event)));
+
+        event = ItemEventFactory.createStateEvent(TEST_ITEM_NAME, DecimalType.ZERO, REMOTE_WEBSOCKET_IMPLEMENTATION);
+        eventWebSocket.processEvent(event);
+        verify(remoteEndpoint).sendString(gson.toJson(new EventDTO(event)));
+
+        // excluded sub-topics are not sent
+        event = ItemEventFactory.createCommandEvent(TEST_ITEM_NAME, DecimalType.ZERO, REMOTE_WEBSOCKET_IMPLEMENTATION);
+        eventWebSocket.processEvent(event);
+        verify(remoteEndpoint, times(2)).sendString(any());
+    }
+
     private void assertEventProcessing(EventDTO incoming, @Nullable Event expectedEvent,
             @Nullable EventDTO expectedResponse) throws IOException {
         eventWebSocket.onText(gson.toJson(incoming));

bundles/org.openhab.core/src/main/java/org/openhab/core/events/TopicEventFilter.java

@@ -12,20 +12,25 @@
  */
 package org.openhab.core.events;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.regex.Pattern;
 
 import org.eclipse.jdt.annotation.NonNullByDefault;
+import org.eclipse.jdt.annotation.Nullable;
 
 /**
  * The {@link TopicEventFilter} is a default openHAB {@link EventFilter} implementation that ensures filtering
- * of events based on an event topic.
+ * of events based on a single event topic or multiple event topics.
  *
  * @author Stefan Bußweiler - Initial contribution
+ * @author Florian Hotze - Add support for filtering of events by multiple event topics
  */
 @NonNullByDefault
 public class TopicEventFilter implements EventFilter {
 
-    private final Pattern topicRegex;
+    private final @Nullable Pattern topicRegex;
+    private final List<Pattern> topicsRegexes = new ArrayList<>();
 
     /**
      * Constructs a new topic event filter.
@@ -38,8 +43,28 @@ public class TopicEventFilter implements EventFilter {
         this.topicRegex = Pattern.compile(topicRegex);
     }
 
+    /**
+     * Constructs a new topic event filter.
+     *
+     * @param topicsRegexes the regular expressions of multiple topics
+     * @see <a href="https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/regex/Pattern.html">Java
+     *      Regex</a>
+     */
+    public TopicEventFilter(List<String> topicsRegexes) {
+        this.topicRegex = null;
+        for (String topicRegex : topicsRegexes) {
+            this.topicsRegexes.add(Pattern.compile(topicRegex));
+        }
+    }
+
     @Override
     public boolean apply(Event event) {
-        return topicRegex.matcher(event.getTopic()).matches();
+        String topic = event.getTopic();
+        Pattern topicRegex = this.topicRegex;
+        if (topicRegex != null) {
+            return topicRegex.matcher(topic).matches();
+        } else {
+            return topicsRegexes.stream().anyMatch(p -> p.matcher(topic).matches());
+        }
     }
 }

bundles/org.openhab.core/src/test/java/org/openhab/core/events/TopicEventFilterTest.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright (c) 2010-2025 Contributors to the openHAB project
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.openhab.core.events;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+
+import org.eclipse.jdt.annotation.NonNullByDefault;
+import org.junit.jupiter.api.Test;
+
+/**
+ * {@link TopicEventFilterTest} tests the {@link TopicEventFilter}.
+ *
+ * @author Florian Hotze - Initial contribution
+ */
+@NonNullByDefault
+public class TopicEventFilterTest {
+    public Event createEvent(String topic) {
+        Event event = mock(Event.class);
+        when(event.getTopic()).thenReturn(topic);
+        return event;
+    }
+
+    @Test
+    public void testSingle() {
+        EventFilter filter = new TopicEventFilter("openhab/items/.*/.*");
+        assertTrue(filter.apply(createEvent("openhab/items/test/command")));
+        assertFalse(filter.apply(createEvent("somewhereElse")));
+        assertFalse(filter.apply(createEvent("preopenhab/items/test/state")));
+
+        filter = new TopicEventFilter("openhab/items/test/command");
+        assertTrue(filter.apply(createEvent("openhab/items/test/command")));
+        assertFalse(filter.apply(createEvent("openhab/items/test/state")));
+    }
+
+    @Test
+    public void testMultiple() {
+        EventFilter filter = new TopicEventFilter(List.of("openhab/items/.*/.*", "openhab/things/.*/.*"));
+        assertTrue(filter.apply(createEvent("openhab/items/test/command")));
+        assertTrue(filter.apply(createEvent("openhab/things/test/added")));
+        assertFalse(filter.apply(createEvent("somewhereElse")));
+        assertFalse(filter.apply(createEvent("preopenhab/items/test/command")));
+
+        filter = new TopicEventFilter(List.of("openhab/items/test/command", "openhab/things/test/added"));
+        assertTrue(filter.apply(createEvent("openhab/items/test/command")));
+        assertTrue(filter.apply(createEvent("openhab/things/test/added")));
+        assertFalse(filter.apply(createEvent("openhab/items/test/state")));
+        assertFalse(filter.apply(createEvent("openhab/things/test/removed")));
+    }
+}