refactor(store): extract shared AbstractEventStore and registry utilities

netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/AbstractEventStore.java

@@ -0,0 +1,98 @@
+/**
+ * Copyright (c) 2025 The Socketio4j Project
+ * Parent project : Copyright (c) 2012-2025 Nikita Koksharov
+ *
+ * Licensed 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 com.socketio4j.socketio.store.event;
+
+import java.util.Arrays;
+
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Common state and channel naming shared by all broker backed {@link EventStore} implementations.
+ */
+public abstract class AbstractEventStore implements EventStore {
+
+    protected final Long nodeId;
+    protected final EventStoreMode eventStoreMode;
+    protected final String channelPrefix;
+
+    protected AbstractEventStore(@Nullable Long nodeId,
+                                 @Nullable EventStoreMode eventStoreMode,
+                                 EventStoreMode defaultEventStoreMode,
+                                 @Nullable String channelPrefix,
+                                 String defaultChannelPrefix) {
+        if (nodeId == null) {
+            nodeId = getNodeId();
+        }
+        this.nodeId = nodeId;
+
+        if (eventStoreMode == null) {
+            eventStoreMode = defaultEventStoreMode;
+        }
+        this.eventStoreMode = eventStoreMode;
+
+        if (channelPrefix == null || channelPrefix.isEmpty()) {
+            channelPrefix = defaultChannelPrefix;
+        }
+        this.channelPrefix = channelPrefix;
+    }
+
+    @Override
+    public EventStoreMode getEventStoreMode() {
+        return eventStoreMode;
+    }
+
+    /**
+     * Maps the event type onto the type actually used for the broker channel:
+     * every type collapses to {@link EventType#ALL_SINGLE_CHANNEL} in single channel mode.
+     */
+    protected EventType resolveType(EventType type) {
+        if (EventStoreMode.SINGLE_CHANNEL.equals(eventStoreMode)) {
+            return EventType.ALL_SINGLE_CHANNEL;
+        }
+        return type;
+    }
+
+    protected String channelName(EventType type) {
+        return channelPrefix + resolveType(type).name();
+    }
+
+    protected void stampNodeId(EventMessage msg) {
+        msg.setNodeId(nodeId);
+    }
+
+    /**
+     * @return true when the message originates from another node and must be dispatched locally.
+     */
+    protected boolean isRemote(EventMessage msg) {
+        return msg != null && !nodeId.equals(msg.getNodeId());
+    }
+
+    protected void unsubscribeAll() {
+        Arrays.stream(EventType.values()).forEach(this::unsubscribe);
+    }
+
+    protected void validateSubscribe(EventType type) {
+        if (EventStoreMode.SINGLE_CHANNEL.equals(eventStoreMode) && type != EventType.ALL_SINGLE_CHANNEL) {
+            throw new UnsupportedOperationException(
+                    "Only ALL_SINGLE_CHANNEL allowed in SINGLE_CHANNEL mode");
+        }
+        if (EventStoreMode.MULTI_CHANNEL.equals(eventStoreMode) && type == EventType.ALL_SINGLE_CHANNEL) {
+            throw new UnsupportedOperationException(
+                    "ALL_SINGLE_CHANNEL not allowed in MULTI_CHANNEL mode");
+        }
+    }
+}

netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/ListenerRegistry.java

@@ -0,0 +1,71 @@
+/**
+ * Copyright (c) 2025 The Socketio4j Project
+ * Parent project : Copyright (c) 2012-2025 Nikita Koksharov
+ *
+ * Licensed 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 com.socketio4j.socketio.store.event;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * Holds the local listeners of poll based event stores and dispatches messages to them.
+ */
+public final class ListenerRegistry {
+
+    private final ConcurrentMap<EventType, Queue<ListenerRegistration<? extends EventMessage>>> listeners =
+            new ConcurrentHashMap<>();
+
+    public <T extends EventMessage> ListenerRegistration<T> register(EventType type,
+                                                                     EventListener<T> listener,
+                                                                     Class<T> clazz) {
+        ListenerRegistration<T> registration = new ListenerRegistration<>(listener, clazz);
+        listeners.computeIfAbsent(type, k -> new ConcurrentLinkedQueue<>()).add(registration);
+        return registration;
+    }
+
+    public void unregister(EventType type, ListenerRegistration<? extends EventMessage> registration) {
+        Queue<ListenerRegistration<? extends EventMessage>> queue = listeners.get(type);
+        if (queue != null) {
+            queue.remove(registration);
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    public <T extends EventMessage> void dispatch(EventType type, EventMessage msg) {
+        Queue<ListenerRegistration<? extends EventMessage>> registrations = listeners.get(type);
+        if (registrations == null) {
+            return;
+        }
+        for (ListenerRegistration<? extends EventMessage> registration : registrations) {
+            if (registration.getClazz().isInstance(msg)) {
+                ((ListenerRegistration<T>) registration).getListener().onMessage((T) msg);
+            }
+        }
+    }
+
+    public void remove(EventType type) {
+        listeners.remove(type);
+    }
+
+    public boolean isEmpty() {
+        return listeners.isEmpty();
+    }
+
+    public void clear() {
+        listeners.clear();
+    }
+}

netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/SubscriptionRegistry.java

@@ -0,0 +1,72 @@
+/**
+ * Copyright (c) 2025 The Socketio4j Project
+ * Parent project : Copyright (c) 2012-2025 Nikita Koksharov
+ *
+ * Licensed 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 com.socketio4j.socketio.store.event;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentMap;
+import java.util.function.BiConsumer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tracks broker subscriptions per {@link EventType} so they can be cancelled on unsubscribe.
+ *
+ * @param <I> registration id returned by the broker client
+ * @param <S> broker handle needed to cancel the registration
+ */
+public final class SubscriptionRegistry<I, S> {
+
+    private static final Logger log = LoggerFactory.getLogger(SubscriptionRegistry.class);
+
+    private final ConcurrentMap<EventType, Queue<I>> registrationIds = new ConcurrentHashMap<>();
+    private final ConcurrentMap<I, S> subscriptions = new ConcurrentHashMap<>();
+
+    public void add(EventType type, I registrationId, S subscription) {
+        subscriptions.put(registrationId, subscription);
+        registrationIds.computeIfAbsent(type, k -> new ConcurrentLinkedQueue<>()).add(registrationId);
+    }
+
+    /**
+     * Removes every registration of the given type, invoking {@code canceller} for each one.
+     * Cancellation failures are logged and never abort the remaining removals.
+     */
+    public void remove(EventType type, BiConsumer<I, S> canceller) {
+        Queue<I> ids = registrationIds.remove(type);
+        if (ids == null || ids.isEmpty()) {
+            return;
+        }
+        for (I id : ids) {
+            S subscription = subscriptions.remove(id);
+            if (subscription == null) {
+                continue;
+            }
+            try {
+                canceller.accept(id, subscription);
+            } catch (Exception ex) {
+                log.warn("Failed to remove subscription {} of type {}", id, type, ex);
+            }
+        }
+    }
+
+    public void clear() {
+        registrationIds.clear();
+        subscriptions.clear();
+    }
+}

netty-socketio-core/src/main/java/com/socketio4j/socketio/store/hazelcast/HazelcastPubSubEventStore.java

@@ -16,42 +16,32 @@
  */
 package com.socketio4j.socketio.store.hazelcast;
 
-import java.util.Arrays;
 import java.util.Objects;
-import java.util.Queue;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ConcurrentMap;
 
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.topic.ITopic;
+import com.socketio4j.socketio.store.event.AbstractEventStore;
 import com.socketio4j.socketio.store.event.EventListener;
 import com.socketio4j.socketio.store.event.EventMessage;
-import com.socketio4j.socketio.store.event.EventStore;
 import com.socketio4j.socketio.store.event.EventStoreMode;
 import com.socketio4j.socketio.store.event.EventType;
+import com.socketio4j.socketio.store.event.SubscriptionRegistry;
 
 
-public class HazelcastPubSubEventStore implements EventStore {
+public class HazelcastPubSubEventStore extends AbstractEventStore {
 
     private final HazelcastInstance hazelcastPub;
     private final HazelcastInstance hazelcastSub;
-    private final Long nodeId;
-    private final EventStoreMode eventStoreMode;
-    private final String topicPrefix;
     private static final String DEFAULT_TOPIC_NAME_PREFIX = "SOCKETIO4J:";
 
-    private final ConcurrentMap<EventType, Queue<UUID>> listenerMap = new ConcurrentHashMap<>();
+    private final SubscriptionRegistry<UUID, ITopic<?>> subscriptions = new SubscriptionRegistry<>();
     private final ConcurrentMap<EventType, ITopic<EventMessage>> activePubTopics = new ConcurrentHashMap<>();
-    private final ConcurrentMap<UUID, ITopic<?>> activeSubTopics = new ConcurrentHashMap<>();
-
-    private static final Logger log = LoggerFactory.getLogger(HazelcastPubSubEventStore.class);
 
     public HazelcastPubSubEventStore(
             @NotNull HazelcastInstance hazelcastPub,
@@ -60,91 +50,43 @@ public HazelcastPubSubEventStore(
             @Nullable EventStoreMode eventStoreMode,
             @Nullable String topicPrefix
     ) {
-        Objects.requireNonNull(hazelcastPub, "hazelcastPub cannot be null");
-        Objects.requireNonNull(hazelcastSub, "hazelcastSub cannot be null");
-
-        if (topicPrefix == null || topicPrefix.isEmpty()) {
-            topicPrefix = DEFAULT_TOPIC_NAME_PREFIX;
-        }
-        this.topicPrefix = topicPrefix;
-
-        if (eventStoreMode == null) {
-            eventStoreMode = EventStoreMode.MULTI_CHANNEL;
-        }
-        this.eventStoreMode = eventStoreMode;
-
-        this.hazelcastPub = hazelcastPub;
-        this.hazelcastSub = hazelcastSub;
-        if (nodeId == null) {
-            nodeId = getNodeId();
-        }
-        this.nodeId = nodeId;
-
+        super(nodeId, eventStoreMode, EventStoreMode.MULTI_CHANNEL, topicPrefix, DEFAULT_TOPIC_NAME_PREFIX);
+        this.hazelcastPub = Objects.requireNonNull(hazelcastPub, "hazelcastPub cannot be null");
+        this.hazelcastSub = Objects.requireNonNull(hazelcastSub, "hazelcastSub cannot be null");
     }
 
     @Override
     public void publish0(EventType type, EventMessage msg) {
-        msg.setNodeId(nodeId);
+        stampNodeId(msg);
 
-        ITopic<EventMessage> topic = activePubTopics.computeIfAbsent(type, k -> {
-            String topicName = getTopicName(k);
-            return hazelcastPub.getTopic(topicName);
-        });
+        ITopic<EventMessage> topic = activePubTopics.computeIfAbsent(type, k -> hazelcastPub.getTopic(channelName(k)));
 
         topic.publish(msg);
     }
-    private String getTopicName(EventType type) {
-        if (EventStoreMode.SINGLE_CHANNEL.equals(eventStoreMode)) {
-            return topicPrefix + EventType.ALL_SINGLE_CHANNEL.name();
-        }
-        return topicPrefix + type.name();
-    }
-    @Override
-    public EventStoreMode getEventStoreMode(){
-        return eventStoreMode;
-    }
 
     @Override
     public <T extends EventMessage> void subscribe0(EventType type, final EventListener<T> listener, Class<T> clazz) {
 
-        ITopic<T> topic = hazelcastSub.getTopic(getTopicName(type));
+        ITopic<T> topic = hazelcastSub.getTopic(channelName(type));
 
         UUID regId = topic.addMessageListener(msg -> {
-            if (!nodeId.equals(msg.getMessageObject().getNodeId())) {
+            if (isRemote(msg.getMessageObject())) {
                 listener.onMessage(msg.getMessageObject());
             }
         });
-        activeSubTopics.put(regId, topic);
 
-        listenerMap.computeIfAbsent(type, k -> new ConcurrentLinkedQueue<>())
-                .add(regId);
+        subscriptions.add(type, regId, topic);
     }
 
     @Override
     public void unsubscribe0(EventType type) {
-        Queue<UUID> regIds = listenerMap.remove(type);
-        if (regIds == null || regIds.isEmpty()) {
-            return;
-        }
-
-        for (UUID id : regIds) {
-            ITopic<?> topic = activeSubTopics.remove(id);
-            if (topic == null) {
-                continue;
-            }
-            try {
-                topic.removeMessageListener(id);
-            } catch (Exception ex) {
-                log.warn("Failed to remove listener {} from topic {}", id, getTopicName(type), ex);
-            }
-        }
+        subscriptions.remove(type, (id, topic) -> topic.removeMessageListener(id));
     }
 
     @Override
     public void shutdown0() {
-        Arrays.stream(EventType.values()).forEach(this::unsubscribe);
-        listenerMap.clear();
-        activeSubTopics.clear();
+        unsubscribeAll();
+        subscriptions.clear();
         activePubTopics.clear();
         //do not shut down client here
     }