From 3ded0cfa67fa2a74f41f830612e95a23d8a17dc2 Mon Sep 17 00:00:00 2001 From: sanjomo Date: Sat, 25 Jul 2026 12:04:54 +0000 Subject: [PATCH] refactor(store): extract shared AbstractEventStore and registries --- .../store/event/AbstractEventStore.java | 98 ++++++++++++++++ .../store/event/ListenerRegistry.java | 71 +++++++++++ .../store/event/SubscriptionRegistry.java | 72 ++++++++++++ .../hazelcast/HazelcastPubSubEventStore.java | 88 +++----------- .../HazelcastPubSubRingBufferEventStore.java | 92 +++------------ .../socketio/store/kafka/KafkaEventStore.java | 110 +++--------------- .../store/nats_pubsub/NatsEventStore.java | 89 +++----------- .../redis_pubsub/RedisPubSubEventStore.java | 80 +++---------- .../RedisPubSubReliableEventStore.java | 92 +++------------ .../redis_stream/RedisStreamEventStore.java | 101 +++------------- 10 files changed, 353 insertions(+), 540 deletions(-) create mode 100644 netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/AbstractEventStore.java create mode 100644 netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/ListenerRegistry.java create mode 100644 netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/SubscriptionRegistry.java diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/AbstractEventStore.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/AbstractEventStore.java new file mode 100644 index 00000000..bfeec71e --- /dev/null +++ b/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"); + } + } +} diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/ListenerRegistry.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/ListenerRegistry.java new file mode 100644 index 00000000..14bb0eaa --- /dev/null +++ b/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>> listeners = + new ConcurrentHashMap<>(); + + public ListenerRegistration register(EventType type, + EventListener listener, + Class clazz) { + ListenerRegistration registration = new ListenerRegistration<>(listener, clazz); + listeners.computeIfAbsent(type, k -> new ConcurrentLinkedQueue<>()).add(registration); + return registration; + } + + public void unregister(EventType type, ListenerRegistration registration) { + Queue> queue = listeners.get(type); + if (queue != null) { + queue.remove(registration); + } + } + + @SuppressWarnings("unchecked") + public void dispatch(EventType type, EventMessage msg) { + Queue> registrations = listeners.get(type); + if (registrations == null) { + return; + } + for (ListenerRegistration registration : registrations) { + if (registration.getClazz().isInstance(msg)) { + ((ListenerRegistration) registration).getListener().onMessage((T) msg); + } + } + } + + public void remove(EventType type) { + listeners.remove(type); + } + + public boolean isEmpty() { + return listeners.isEmpty(); + } + + public void clear() { + listeners.clear(); + } +} diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/SubscriptionRegistry.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/SubscriptionRegistry.java new file mode 100644 index 00000000..3d67aa97 --- /dev/null +++ b/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 registration id returned by the broker client + * @param broker handle needed to cancel the registration + */ +public final class SubscriptionRegistry { + + private static final Logger log = LoggerFactory.getLogger(SubscriptionRegistry.class); + + private final ConcurrentMap> registrationIds = new ConcurrentHashMap<>(); + private final ConcurrentMap 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 canceller) { + Queue 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(); + } +} diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/hazelcast/HazelcastPubSubEventStore.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/hazelcast/HazelcastPubSubEventStore.java index 6490c3b2..256e634b 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/hazelcast/HazelcastPubSubEventStore.java +++ b/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> listenerMap = new ConcurrentHashMap<>(); + private final SubscriptionRegistry> subscriptions = new SubscriptionRegistry<>(); private final ConcurrentMap> activePubTopics = new ConcurrentHashMap<>(); - private final ConcurrentMap> 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 topic = activePubTopics.computeIfAbsent(type, k -> { - String topicName = getTopicName(k); - return hazelcastPub.getTopic(topicName); - }); + ITopic 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 void subscribe0(EventType type, final EventListener listener, Class clazz) { - ITopic topic = hazelcastSub.getTopic(getTopicName(type)); + ITopic 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 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 } diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/hazelcast_ringbuffer/HazelcastPubSubRingBufferEventStore.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/hazelcast_ringbuffer/HazelcastPubSubRingBufferEventStore.java index 71ccaa8a..c1f334a2 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/hazelcast_ringbuffer/HazelcastPubSubRingBufferEventStore.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/hazelcast_ringbuffer/HazelcastPubSubRingBufferEventStore.java @@ -16,42 +16,32 @@ */ package com.socketio4j.socketio.store.hazelcast_ringbuffer; -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.EventStoreType; import com.socketio4j.socketio.store.event.EventType; +import com.socketio4j.socketio.store.event.SubscriptionRegistry; -public class HazelcastPubSubRingBufferEventStore implements EventStore { +public class HazelcastPubSubRingBufferEventStore extends AbstractEventStore { private final HazelcastInstance hazelcastPub; private final HazelcastInstance hazelcastSub; - private final Long nodeId; - private final EventStoreMode eventStoreMode; - private final ConcurrentMap> listenerMap = new ConcurrentHashMap<>(); + private final SubscriptionRegistry> subscriptions = new SubscriptionRegistry<>(); private final ConcurrentMap> activePubTopics = new ConcurrentHashMap<>(); - private final ConcurrentMap> activeSubTopics = new ConcurrentHashMap<>(); - - private static final Logger log = LoggerFactory.getLogger(HazelcastPubSubRingBufferEventStore.class); - private final String ringBufferNamePrefix; private static final String DEFAULT_RING_BUFFER_NAME_PREFIX = "SOCKETIO4J:"; @@ -62,51 +52,22 @@ public HazelcastPubSubRingBufferEventStore( @Nullable EventStoreMode eventStoreMode, @Nullable String ringBufferNamePrefix ) { - Objects.requireNonNull(hazelcastPub, "hazelcastPub cannot be null"); - Objects.requireNonNull(hazelcastSub, "hazelcastSub cannot be null"); - - if (ringBufferNamePrefix == null || ringBufferNamePrefix.isEmpty()) { - ringBufferNamePrefix = DEFAULT_RING_BUFFER_NAME_PREFIX; - } - this.ringBufferNamePrefix = ringBufferNamePrefix; - - 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, + ringBufferNamePrefix, DEFAULT_RING_BUFFER_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 topic = activePubTopics.computeIfAbsent(type, k -> { - String topicName = getRingBufferName(k); - return hazelcastPub.getReliableTopic(topicName); - }); + ITopic topic = activePubTopics.computeIfAbsent( + type, k -> hazelcastPub.getReliableTopic(channelName(k))); topic.publish(msg); } - private String getRingBufferName(EventType type) { - if (EventStoreMode.SINGLE_CHANNEL.equals(eventStoreMode)) { - return ringBufferNamePrefix + EventType.ALL_SINGLE_CHANNEL.name(); - } - return ringBufferNamePrefix + type.name(); - } - - @Override - public EventStoreMode getEventStoreMode(){ - return eventStoreMode; - } @Override public EventStoreType getEventStoreType() { @@ -116,45 +77,26 @@ public EventStoreType getEventStoreType() { @Override public void subscribe0(EventType type, final EventListener listener, Class clazz) { - ITopic topic = hazelcastSub.getReliableTopic(getRingBufferName(type)); + ITopic topic = hazelcastSub.getReliableTopic(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 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, getRingBufferName(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 } diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/kafka/KafkaEventStore.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/kafka/KafkaEventStore.java index 383041cf..91014445 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/kafka/KafkaEventStore.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/kafka/KafkaEventStore.java @@ -21,10 +21,8 @@ import java.util.List; import java.util.Objects; import java.util.Properties; -import java.util.Queue; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; @@ -48,20 +46,21 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +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.EventStoreType; import com.socketio4j.socketio.store.event.EventType; import com.socketio4j.socketio.store.event.ListenerRegistration; +import com.socketio4j.socketio.store.event.ListenerRegistry; /** * @author https://github.com/sanjomo * @date 15/12/25 6:09 pm */ -public final class KafkaEventStore implements EventStore { +public final class KafkaEventStore extends AbstractEventStore { private static final Logger log = LoggerFactory.getLogger(KafkaEventStore.class); @@ -72,9 +71,8 @@ public final class KafkaEventStore implements EventStore { private final KafkaProducer producer; private final Properties consumerProps; - private final String topicPrefix; - private final Long nodeId; - private final EventStoreMode mode; + + private static final String DEFAULT_TOPIC_PREFIX = "SOCKETIO4J-"; // --------------------------------------------------------------------- // Runtime @@ -88,8 +86,7 @@ public final class KafkaEventStore implements EventStore { private final ConcurrentMap pollers = new ConcurrentHashMap<>(); - private final ConcurrentMap>> listeners = - new ConcurrentHashMap<>(); + private final ListenerRegistry listeners = new ListenerRegistry(); /** * Completes after {@link #createConsumer} finishes (assign + seek) for that poller. @@ -111,33 +108,16 @@ public KafkaEventStore( @Nullable EventStoreMode mode, @Nullable String topicPrefix ) { + super(nodeId, mode, EventStoreMode.MULTI_CHANNEL, topicPrefix, DEFAULT_TOPIC_PREFIX); this.producer = Objects.requireNonNull(producer); this.consumerProps = Objects.requireNonNull(consumerProps); - - if (nodeId == null){ - nodeId = getNodeId(); - } - this.nodeId = Objects.requireNonNull(nodeId); - if (mode == null) { - mode = EventStoreMode.MULTI_CHANNEL; - } - this.mode = Objects.requireNonNull(mode); - if (topicPrefix == null || topicPrefix.isEmpty()) { - topicPrefix = "SOCKETIO4J-"; - } - this.topicPrefix = Objects.requireNonNull(topicPrefix); } // --------------------------------------------------------------------- // Metadata // --------------------------------------------------------------------- - @Override - public EventStoreMode getEventStoreMode() { - return mode; - } - @Override public EventStoreType getEventStoreType() { return EventStoreType.STREAM; @@ -166,9 +146,9 @@ public EventStoreType getEventStoreType() { @Override public void publish0(EventType type, EventMessage msg) { - msg.setNodeId(nodeId); + stampNodeId(msg); - String topic = topic(resolve(type)); + String topic = channelName(type); ProducerRecord record = new ProducerRecord<>(topic, type.name(), msg); @@ -202,18 +182,14 @@ public void subscribe0( validateSubscribe(type); - ListenerRegistration registration = - new ListenerRegistration<>(listener, clazz); - Queue> queue = - listeners.computeIfAbsent(type, k -> new ConcurrentLinkedQueue<>()); - queue.add(registration); + ListenerRegistration registration = listeners.register(type, listener, clazz); boolean consumerReady = false; try { ensureConsumer(type); consumerReady = true; } finally { if (!consumerReady) { - queue.remove(registration); + listeners.unregister(type, registration); } } } @@ -319,7 +295,7 @@ private KafkaConsumer createConsumer(EventType type) { KafkaConsumer consumer = new KafkaConsumer<>(props); - String topic = topic(resolve(type)); + String topic = channelName(type); // Discover partitions with retry logic // Topics may be auto-created by Kafka, so we retry a few times @@ -412,7 +388,7 @@ private void pollLoop(EventType type, EventMessage msg = rec.value(); // Skip null messages and messages from this node (already processed locally) - if (msg == null || nodeId.equals(msg.getNodeId())) { + if (!isRemote(msg)) { continue; } @@ -462,25 +438,11 @@ private void pollLoop(EventType type, // --------------------------------------------------------------------- - private void dispatch( + private void dispatch( EventType type, EventMessage msg ) { - - Queue> regs = - listeners.get(type); - - if (regs == null) { - return; - } - - for (ListenerRegistration reg : regs) { - if (reg.getClazz().isInstance(msg)) { - ((ListenerRegistration) reg) - .getListener() - .onMessage((T) msg); - } - } + listeners.dispatch(type, msg); } // --------------------------------------------------------------------- @@ -534,46 +496,4 @@ public void shutdown0() { // Utils // --------------------------------------------------------------------- - private String topic(EventType type) { - return topicPrefix + type.name(); - } - - /** - * Resolves the event type based on the store mode. - * - *

In SINGLE_CHANNEL mode, all events are routed to a single topic (ALL_SINGLE_CHANNEL). - * This ensures event ordering across all event types but requires all nodes to process - * all events. - * - *

In MULTI_CHANNEL mode, each event type has its own topic, allowing independent - * scaling and processing of different event types. - * - * @param type the original event type - * @return the resolved event type (may be ALL_SINGLE_CHANNEL in single channel mode) - */ - private EventType resolve(EventType type) { - if (mode == EventStoreMode.SINGLE_CHANNEL) { - return EventType.ALL_SINGLE_CHANNEL; - } - return type; - } - - /** - * Validates that the subscription request is compatible with the current store mode. - * - * @param type the event type to subscribe to - * @throws UnsupportedOperationException if the subscription is invalid for the current mode - */ - private void validateSubscribe(EventType type) { - - if (mode == EventStoreMode.SINGLE_CHANNEL && type != EventType.ALL_SINGLE_CHANNEL) { - throw new UnsupportedOperationException( - "Only ALL_SINGLE_CHANNEL allowed in SINGLE_CHANNEL mode"); - } - - if (mode == EventStoreMode.MULTI_CHANNEL && type == EventType.ALL_SINGLE_CHANNEL) { - throw new UnsupportedOperationException( - "ALL_SINGLE_CHANNEL not allowed in MULTI_CHANNEL mode"); - } - } } diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/nats_pubsub/NatsEventStore.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/nats_pubsub/NatsEventStore.java index 47f1bace..920475ab 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/nats_pubsub/NatsEventStore.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/nats_pubsub/NatsEventStore.java @@ -16,23 +16,19 @@ */ package com.socketio4j.socketio.store.nats_pubsub; -import java.util.Arrays; import java.util.Objects; -import java.util.Queue; -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.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; import io.nats.client.Connection; import io.nats.client.Dispatcher; @@ -45,26 +41,14 @@ * Unreliable NATS Core based EventStore. * Events are ephemeral and not replayed. */ -public class NatsEventStore implements EventStore { +public class NatsEventStore extends AbstractEventStore { private static final Logger log = LoggerFactory.getLogger(NatsEventStore.class); private final Connection nats; - private final Long nodeId; - private final EventStoreMode eventStoreMode; - /** - * EventType -> subscriptions - */ - private final ConcurrentMap> subscriptions = - new ConcurrentHashMap<>(); - - /** - * Subscription -> dispatcher - */ - private final ConcurrentMap activeDispatchers = - new ConcurrentHashMap<>(); + private final SubscriptionRegistry subscriptions = new SubscriptionRegistry<>(); // ---------------------------------------------------------------------- // Constructors @@ -80,36 +64,21 @@ public class NatsEventStore implements EventStore { public NatsEventStore(@NotNull Connection natsConnection, @Nullable EventStoreMode eventStoreMode, @Nullable Long nodeId) { - + super(nodeId, eventStoreMode, EventStoreMode.MULTI_CHANNEL, null, ""); this.nats = Objects.requireNonNull(natsConnection, "natsConnection"); - - if (nodeId == null) { - nodeId = getNodeId(); - } - this.nodeId = nodeId; - - if (eventStoreMode == null) { - eventStoreMode = EventStoreMode.MULTI_CHANNEL; - } - this.eventStoreMode = eventStoreMode; } // ---------------------------------------------------------------------- // EventStore SPI // ---------------------------------------------------------------------- - @Override - public EventStoreMode getEventStoreMode() { - return eventStoreMode; - } - @Override public void publish0(EventType type, EventMessage msg) { - msg.setNodeId(nodeId); + stampNodeId(msg); try { byte[] data = EventMessageCodec.serialize(msg); - nats.publish(getSubjectName(type), data); + nats.publish(channelName(type), data); } catch (Exception e) { log.warn("Failed to publish event {}", type, e); } @@ -121,13 +90,13 @@ public void subscribe0( final EventListener listener, Class clazz) { - final String subject = getSubjectName(type); + final String subject = channelName(type); final Dispatcher dispatcher = nats.createDispatcher(); Subscription subscription = dispatcher.subscribe(subject, (Message msg) -> { try { T event = EventMessageCodec.deserialize(msg.getData(), clazz); - if (!nodeId.equals(event.getNodeId())) { + if (isRemote(event)) { listener.onMessage(event); } } catch (Exception e) { @@ -135,49 +104,21 @@ public void subscribe0( } }); - activeDispatchers.put(subscription, dispatcher); - subscriptions - .computeIfAbsent(type, k -> new ConcurrentLinkedQueue<>()) - .add(subscription); + subscriptions.add(type, subscription, dispatcher); } @Override public void unsubscribe0(EventType type) { - Queue subs = subscriptions.remove(type); - if (subs == null || subs.isEmpty()) { - return; - } - - for (Subscription sub : subs) { - try { - Dispatcher dispatcher = activeDispatchers.remove(sub); - if (dispatcher != null) { - dispatcher.unsubscribe(sub); - //sub.unsubscribe(); - nats.closeDispatcher(dispatcher); - } - } catch (Exception e) { - log.warn("Failed to unsubscribe from {}", type, e); - } - } + subscriptions.remove(type, (subscription, dispatcher) -> { + dispatcher.unsubscribe(subscription); + nats.closeDispatcher(dispatcher); + }); } @Override public void shutdown0() { - Arrays.stream(EventType.values()).forEach(this::unsubscribe); + unsubscribeAll(); subscriptions.clear(); - activeDispatchers.clear(); - } - - // ---------------------------------------------------------------------- - // Helpers - // ---------------------------------------------------------------------- - - private String getSubjectName(EventType type) { - if (EventStoreMode.SINGLE_CHANNEL.equals(eventStoreMode)) { - return EventType.ALL_SINGLE_CHANNEL.name(); - } - return type.name(); } // ---------------------------------------------------------------------- diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_pubsub/RedisPubSubEventStore.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_pubsub/RedisPubSubEventStore.java index 0ff7861b..98182bae 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_pubsub/RedisPubSubEventStore.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_pubsub/RedisPubSubEventStore.java @@ -16,43 +16,34 @@ */ package com.socketio4j.socketio.store.redis_pubsub; -import java.util.Arrays; import java.util.Objects; -import java.util.Queue; 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.redisson.api.RTopic; import org.redisson.api.RedissonClient; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +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; /** * Unreliable Redis Pub/Sub based EventStore. * Events are ephemeral and not replayed. */ -public class RedisPubSubEventStore implements EventStore { +public class RedisPubSubEventStore extends AbstractEventStore { private final RedissonClient redissonPub; private final RedissonClient redissonSub; - private final Long nodeId; - private final EventStoreMode eventStoreMode; - private final ConcurrentMap> map = new ConcurrentHashMap<>(); - private final ConcurrentMap activeSubTopics = new ConcurrentHashMap<>(); + private final SubscriptionRegistry subscriptions = new SubscriptionRegistry<>(); private final ConcurrentMap activePubTopics = new ConcurrentHashMap<>(); - private static final Logger log = LoggerFactory.getLogger(RedisPubSubEventStore.class); - // ---------------------------------------------------------------------- // Constructors // ---------------------------------------------------------------------- @@ -67,78 +58,39 @@ public RedisPubSubEventStore(@NotNull RedissonClient redissonPub, @NotNull RedissonClient redissonSub, @Nullable EventStoreMode eventStoreMode, @Nullable Long nodeId) { - Objects.requireNonNull(redissonPub, "redissonPub is null"); - Objects.requireNonNull(redissonSub, "redissonSub is null"); - - this.redissonPub = redissonPub; - this.redissonSub = redissonSub; - if (nodeId == null) { - nodeId = getNodeId(); - } - this.nodeId = nodeId; - if (eventStoreMode == null) { - eventStoreMode = EventStoreMode.MULTI_CHANNEL; - } - this.eventStoreMode = eventStoreMode; + super(nodeId, eventStoreMode, EventStoreMode.MULTI_CHANNEL, null, ""); + this.redissonPub = Objects.requireNonNull(redissonPub, "redissonPub is null"); + this.redissonSub = Objects.requireNonNull(redissonSub, "redissonSub is null"); } - @Override - public EventStoreMode getEventStoreMode(){ - return this.eventStoreMode; - } @Override public void publish0(EventType type, EventMessage msg) { - msg.setNodeId(nodeId); - RTopic topic = activePubTopics.computeIfAbsent(type, k -> { - String topicName = getStreamName(k); - return redissonPub.getTopic(topicName); - }); + stampNodeId(msg); + RTopic topic = activePubTopics.computeIfAbsent(type, k -> redissonPub.getTopic(channelName(k))); topic.publish(msg); } @Override public void subscribe0(EventType type, final EventListener listener, Class clazz) { - RTopic topic = redissonSub.getTopic(getStreamName(type)); + RTopic topic = redissonSub.getTopic(channelName(type)); int regId = topic.addListener(clazz, (channel, msg) -> { - if (!nodeId.equals(msg.getNodeId())) { + if (isRemote(msg)) { listener.onMessage(msg); } }); - activeSubTopics.put(regId, topic); - map.computeIfAbsent(type, k -> new ConcurrentLinkedQueue<>()).add(regId); - } - private String getStreamName(EventType type) { - if (EventStoreMode.SINGLE_CHANNEL.equals(eventStoreMode)) { - return EventType.ALL_SINGLE_CHANNEL.name(); - } - return type.name(); + subscriptions.add(type, regId, topic); } + @Override public void unsubscribe0(EventType type) { - - Queue regIds = map.remove(type); - if (regIds == null || regIds.isEmpty()) { - return; - } - for (Integer id : regIds) { - RTopic topic = activeSubTopics.remove(id); - if (topic == null) { - continue; - } - try { - topic.removeListener(id); - } catch (Exception ex) { - log.warn("Failed to remove listener {} from topic {}", id, getStreamName(type), ex); - } - } + subscriptions.remove(type, (id, topic) -> topic.removeListener(id)); } @Override public void shutdown0() { - Arrays.stream(EventType.values()).forEach(this::unsubscribe); - map.clear(); + unsubscribeAll(); + subscriptions.clear(); activePubTopics.clear(); - activeSubTopics.clear(); } public static final class Builder { diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_reliable/RedisPubSubReliableEventStore.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_reliable/RedisPubSubReliableEventStore.java index 91a40d12..6d7b4d31 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_reliable/RedisPubSubReliableEventStore.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_reliable/RedisPubSubReliableEventStore.java @@ -17,11 +17,8 @@ package com.socketio4j.socketio.store.redis_reliable; import java.time.Duration; -import java.util.Arrays; import java.util.Objects; -import java.util.Queue; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -36,28 +33,25 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +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.EventStoreType; import com.socketio4j.socketio.store.event.EventType; import com.socketio4j.socketio.store.event.PublishMode; +import com.socketio4j.socketio.store.event.SubscriptionRegistry; -public class RedisPubSubReliableEventStore implements EventStore { +public class RedisPubSubReliableEventStore extends AbstractEventStore { private final RedissonClient redissonPub; private final RedissonClient redissonSub; - private final Long nodeId; - private final EventStoreMode eventStoreMode; - private final String streamNamePrefix; private final Integer streamMaxLength; private final Duration trimEvery; private final ScheduledExecutorService trimExecutor; private static final String DEFAULT_STREAM_NAME_PREFIX = "SOCKETIO4J:"; private static final int DEFAULT_STREAM_MAX_LENGTH = Integer.MAX_VALUE; - private final ConcurrentMap> map = new ConcurrentHashMap<>(); - private final ConcurrentMap activeSubTopics = new ConcurrentHashMap<>(); + private final SubscriptionRegistry subscriptions = new SubscriptionRegistry<>(); private final ConcurrentMap activePubTopics = new ConcurrentHashMap<>(); private final ConcurrentMap> trimTopics = new ConcurrentHashMap<>(); private static final Logger log = LoggerFactory.getLogger(RedisPubSubReliableEventStore.class); @@ -74,21 +68,11 @@ public RedisPubSubReliableEventStore(@NotNull RedissonClient redissonPub, @Nullable String streamNamePrefix, @Nullable Integer streamMaxLength, @Nullable Duration trimEvery) { - - if (eventStoreMode == null) { - eventStoreMode = EventStoreMode.MULTI_CHANNEL; - } - this.eventStoreMode = eventStoreMode; + super(nodeId, eventStoreMode, EventStoreMode.MULTI_CHANNEL, streamNamePrefix, DEFAULT_STREAM_NAME_PREFIX); Objects.requireNonNull(redissonPub, "redissonPub client can not be null"); Objects.requireNonNull(redissonSub, "redissonSub client can not be null"); - if (streamNamePrefix == null || streamNamePrefix.isEmpty()) { - streamNamePrefix = DEFAULT_STREAM_NAME_PREFIX; - log.warn("streamNamePrefix is null/empty, loaded default : {}", DEFAULT_STREAM_NAME_PREFIX); - } - this.streamNamePrefix = streamNamePrefix; - if (streamMaxLength == null || streamMaxLength <=0) { streamMaxLength = DEFAULT_STREAM_MAX_LENGTH; log.warn("streamMaxLength is null/less than 1, loaded default : {}", DEFAULT_STREAM_MAX_LENGTH); @@ -106,10 +90,6 @@ public RedisPubSubReliableEventStore(@NotNull RedissonClient redissonPub, this.redissonPub = redissonPub; this.redissonSub = redissonSub; - if (nodeId == null) { - nodeId = getNodeId(); - } - this.nodeId = nodeId; ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(r -> { Thread t = new Thread(r, "socketio4j-redis-stream-trimmer"); @@ -152,7 +132,7 @@ private void trimAllReliableStreams() { } } private RStream createStream(EventType type) { - return redissonPub.getStream(getStreamName(type)); + return redissonPub.getStream(channelName(type)); } /** @@ -177,7 +157,7 @@ private void trimStream(EventType type) { StreamTrimArgs.maxLen(streamMaxLength).noLimit() ).whenComplete((trimmed, err) -> { if (err != null) { - log.warn("Trim failed for {}", getStreamName(type), err); + log.warn("Trim failed for {}", channelName(type), err); return; } @@ -185,25 +165,20 @@ private void trimStream(EventType type) { stream.sizeAsync() .whenComplete((length, sizeErr) -> { if (sizeErr != null) { - log.warn("Failed to read stream size {}", getStreamName(type), sizeErr); + log.warn("Failed to read stream size {}", channelName(type), sizeErr); return; } - log.debug("Stream {} length={}", getStreamName(type), length); + log.debug("Stream {} length={}", channelName(type), length); }); }); } catch (Exception e) { - log.warn("Failed to trim Redis stream {}", getStreamName(type), e); + log.warn("Failed to trim Redis stream {}", channelName(type), e); } } - @Override - public EventStoreMode getEventStoreMode(){ - return eventStoreMode; - } - @Override public EventStoreType getEventStoreType() { return EventStoreType.STREAM; @@ -215,56 +190,28 @@ public PublishMode getPublishMode(){ } @Override public void publish0(EventType type, EventMessage msg) { - msg.setNodeId(nodeId); - RReliableTopic topic = activePubTopics.computeIfAbsent(type, k -> { - String topicName = getStreamName(k); - return redissonPub.getReliableTopic(topicName); - }); + stampNodeId(msg); + RReliableTopic topic = activePubTopics.computeIfAbsent( + type, k -> redissonPub.getReliableTopic(channelName(k))); topic.publish(msg); } @Override public void subscribe0(EventType type, final EventListener listener, Class clazz) { - RReliableTopic reliableTopic = redissonSub.getReliableTopic(getStreamName(type)); + RReliableTopic reliableTopic = redissonSub.getReliableTopic(channelName(type)); Objects.requireNonNull(reliableTopic, "reliableTopic can not be null"); String id = reliableTopic.addListener(clazz, (channel, msg) -> { - if (!nodeId.equals(msg.getNodeId())) { + if (isRemote(msg)) { listener.onMessage(msg); } }); - activeSubTopics.put(id, reliableTopic); - map.computeIfAbsent(type, k -> new ConcurrentLinkedQueue<>()) - .add(id); - } - - private String getStreamName(EventType type) { - if (EventStoreMode.SINGLE_CHANNEL.equals(getEventStoreMode())) { - return streamNamePrefix + EventType.ALL_SINGLE_CHANNEL.name(); - } - return streamNamePrefix + type.name(); + subscriptions.add(type, id, reliableTopic); } @Override public void unsubscribe0(EventType type) { - - Queue regIds = map.remove(type); - if (regIds == null || regIds.isEmpty()) { - return; - } - for (String id : regIds) { - RReliableTopic topic = activeSubTopics.remove(id); - if (topic == null) { - continue; - } - try { - topic.removeListener(id); - } catch (Exception ex) { - log.warn("Failed to remove listener {} from topic {}", id, getStreamName(type), ex); - } - } - - + subscriptions.remove(type, (id, topic) -> topic.removeListener(id)); } @@ -274,12 +221,11 @@ public void shutdown0() { trimExecutor.shutdown(); // Unsubscribe from all event types - Arrays.stream(EventType.values()).forEach(this::unsubscribe); - map.clear(); + unsubscribeAll(); + subscriptions.clear(); // Clear all topic references activePubTopics.clear(); - activeSubTopics.clear(); trimTopics.clear(); } diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_stream/RedisStreamEventStore.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_stream/RedisStreamEventStore.java index 44b9dad3..d64959d4 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_stream/RedisStreamEventStore.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_stream/RedisStreamEventStore.java @@ -19,9 +19,7 @@ import java.time.Duration; import java.util.Arrays; import java.util.Objects; -import java.util.Queue; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -38,16 +36,16 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +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.EventStoreType; import com.socketio4j.socketio.store.event.EventType; -import com.socketio4j.socketio.store.event.ListenerRegistration; +import com.socketio4j.socketio.store.event.ListenerRegistry; -public class RedisStreamEventStore implements EventStore { +public class RedisStreamEventStore extends AbstractEventStore { private static final Logger log = LoggerFactory.getLogger(RedisStreamEventStore.class); @@ -65,9 +63,6 @@ public class RedisStreamEventStore implements EventStore { private final RedissonClient redissonPub; private final RedissonClient redissonSub; - private final Long nodeId; - private final EventStoreMode eventStoreMode; - private final String streamNamePrefix; private final int streamMaxLength; // --------------------------------------------------------------------- @@ -81,8 +76,7 @@ public class RedisStreamEventStore implements EventStore { private final ConcurrentMap> subStreams = new ConcurrentHashMap<>(); - private final ConcurrentMap>> listeners = - new ConcurrentHashMap<>(); + private final ListenerRegistry listeners = new ListenerRegistry(); private final ConcurrentMap offsets = new ConcurrentHashMap<>(); @@ -102,28 +96,11 @@ public RedisStreamEventStore( @Nullable String streamNamePrefix, @Nullable Integer streamMaxLength ) { + super(nodeId, eventStoreMode, EventStoreMode.SINGLE_CHANNEL, streamNamePrefix, DEFAULT_PREFIX); this.redissonPub = Objects.requireNonNull(redissonPub, "redissonPub"); this.redissonSub = Objects.requireNonNull(redissonSub, "redissonSub"); - if (nodeId == null) { - nodeId = getNodeId(); - log.warn("nodeId is null, loaded default : {}", nodeId); - } - this.nodeId = nodeId; - - if (eventStoreMode == null) { - eventStoreMode = EventStoreMode.SINGLE_CHANNEL; - log.warn("mode is null, loaded default : {}", EventStoreMode.SINGLE_CHANNEL); - } - this.eventStoreMode = eventStoreMode; - - if (streamNamePrefix == null || streamNamePrefix.isEmpty()) { - streamNamePrefix = DEFAULT_PREFIX; - log.warn("prefix is null/empty, loaded default : {}", DEFAULT_PREFIX); - } - this.streamNamePrefix = streamNamePrefix; - if (streamMaxLength == null || streamMaxLength <= 0) { streamMaxLength = DEFAULT_MAX_LEN; log.warn( @@ -153,8 +130,8 @@ private void initStreams() { } private void initStream(EventType type) { - subStreams.put(type, redissonSub.getStream(streamName(type))); - pubStreams.put(type, redissonPub.getStream(streamName(type))); + subStreams.put(type, redissonSub.getStream(channelName(type))); + pubStreams.put(type, redissonPub.getStream(channelName(type))); offsets.put(type, StreamMessageId.NEWEST); } @@ -162,11 +139,6 @@ private void initStream(EventType type) { // Metadata // --------------------------------------------------------------------- - @Override - public EventStoreMode getEventStoreMode() { - return eventStoreMode; - } - @Override public EventStoreType getEventStoreType() { return EventStoreType.STREAM; @@ -178,11 +150,11 @@ public EventStoreType getEventStoreType() { @Override public void publish0(EventType type, EventMessage msg) { - msg.setNodeId(nodeId); + stampNodeId(msg); pubStreams.computeIfAbsent( - resolve(type), - t -> redissonPub.getStream(streamName(t)) + resolveType(type), + t -> redissonPub.getStream(channelName(t)) ).add(StreamAddArgs.entry(type.name(), msg).trimNonStrict().maxLen(streamMaxLength).noLimit()); } @@ -203,9 +175,7 @@ public void subscribe0( validateSubscribe(type); - listeners - .computeIfAbsent(type, k -> new ConcurrentLinkedQueue<>()) - .add(new ListenerRegistration<>(listener, clazz)); + listeners.register(type, listener, clazz); ensurePoller(type); } @@ -226,7 +196,7 @@ private void ensurePoller(EventType type) { RStream stream = subStreams.computeIfAbsent( t, - k -> redissonSub.getStream(streamName(k)) + k -> redissonSub.getStream(channelName(k)) ); newExec.execute(() -> pollLoop(stream, t)); return newExec; @@ -264,7 +234,7 @@ private void pollLoop(RStream stream, EventType type) { EventMessage msg = map.values().iterator().next(); try { - if (!nodeId.equals(msg.getNodeId())) { + if (isRemote(msg)) { dispatch(type, msg, id); } } finally { @@ -280,28 +250,13 @@ private void pollLoop(RStream stream, EventType type) { }); } - private void dispatch( + private void dispatch( EventType type, EventMessage msg, StreamMessageId id ) { - - Queue> regs = - listeners.get(type); - - if (regs == null) { - return; - } - msg.setOffset(id.toString()); - - for (ListenerRegistration reg : regs) { - if (reg.getClazz().isInstance(msg)) { - ((ListenerRegistration) reg) - .getListener() - .onMessage((T) msg); - } - } + listeners.dispatch(type, msg); } private void scheduleRetry(RStream stream, EventType type) { @@ -345,32 +300,6 @@ public void shutdown0() { // Utils // --------------------------------------------------------------------- - private String streamName(EventType type) { - if (EventStoreMode.SINGLE_CHANNEL.equals(eventStoreMode)) { - return streamNamePrefix + EventType.ALL_SINGLE_CHANNEL.name(); - } - return streamNamePrefix + type.name(); - } - - private EventType resolve(EventType type) { - if (EventStoreMode.SINGLE_CHANNEL.equals(eventStoreMode)) { - return EventType.ALL_SINGLE_CHANNEL; - } - return type; - } - - private 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"); - } - } public static final class Builder { // -------------------------