dschneider-pivotal commented on a change in pull request #7408: URL: https://github.com/apache/geode/pull/7408#discussion_r819838911
########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/BlockingCommandListener.java ########## @@ -0,0 +1,99 @@ +/* + * 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.geode.redis.internal.eventing; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.apache.geode.redis.internal.commands.Command; +import org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; +import org.apache.geode.redis.internal.netty.Coder; +import org.apache.geode.redis.internal.netty.ExecutionHandlerContext; + +public class BlockingCommandListener implements EventListener { + + private final ExecutionHandlerContext context; + private final RedisCommandType command; + private final List<RedisKey> keys; + private final byte[][] commandOptions; + private final long timeout; + + /** + * Constructor to create an instance of a BlockingCommandListener in response to a blocking + * command. When receiving a relevant event, blocking commands simply resubmit the command + * into the Netty pipeline. + * + * @param context the associated ExecutionHandlerContext + * @param command the blocking command + * @param timeout the timeout for the command to block + * @param keys the list of keys the command is interested in + * @param commandOptions any additional options (other than the keys and timeout) required to + * resubmit the command. + */ + public BlockingCommandListener(ExecutionHandlerContext context, RedisCommandType command, + long timeout, List<RedisKey> keys, byte[]... commandOptions) { + this.context = context; + this.command = command; + this.timeout = timeout; + this.keys = Collections.unmodifiableList(keys); + this.commandOptions = commandOptions; + } + + @Override + public List<RedisKey> keys() { + return keys; + } + + @Override + public EventResponse process(RedisCommandType commandType, RedisKey key) { + if (!keys.contains(key)) { + return EventResponse.CONTIINUE; + } + + resubmitCommand(); + return EventResponse.REMOVE_AND_STOP; + } + + @Override + public void resubmitCommand() { + List<byte[]> byteArgs = new ArrayList<>(keys.size() + 1); + byteArgs.add(command.name().getBytes()); + + keys.forEach(x -> byteArgs.add(x.toBytes())); + byteArgs.addAll(Arrays.asList(commandOptions)); + + // The commands we are currently supporting all have the timeout at the end of the argument + // list. Some newer Redis 7 commands (BLMPOP and BZMPOP) have the timeout as the first argument + // after the command. We'll need to adjust this once those commands are supported. + byteArgs.add(Coder.longToBytes(timeout)); + + context.resubmitCommand(new Command(command, byteArgs)); Review comment: Could instead have Command.resubmit(context)? Command is immutable so can't we just reuse it instead of creating a new one? And Command knows how to find its RedisCommandType and use it to call commandExecutor.executeCommand. No need to call checkParameters again. It seems like this would let you get rid of commandOptions on this class and no need to rebuild argument list in this method. It would also get rid of the redis 7 issue you comment on. ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/EventDistributor.java ########## @@ -0,0 +1,128 @@ +/* + * 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.geode.redis.internal.eventing; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.Timer; +import java.util.TimerTask; + +import org.apache.logging.log4j.Logger; + +import org.apache.geode.annotations.VisibleForTesting; +import org.apache.geode.cache.partition.PartitionListenerAdapter; +import org.apache.geode.logging.internal.log4j.api.LogService; +import org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; + +public class EventDistributor extends PartitionListenerAdapter { + + private static final Logger logger = LogService.getLogger(); + private final Map<RedisKey, List<EventListener>> listeners = new HashMap<>(); + private final Map<EventListener, TimerTask> timerTasks = new HashMap<>(); + private final Timer timer = new Timer("GeodeForRedisEventTimer", true); + private int keysRegistered = 0; + + public synchronized void registerListener(EventListener listener) { + for (RedisKey key : listener.keys()) { + listeners.computeIfAbsent(key, k -> new ArrayList<>()).add(listener); + } + + if (listener.getTimeout() != 0) { + scheduleTimeout(listener, listener.getTimeout()); + } + + keysRegistered += listener.keys().size(); + } + + public synchronized void fireEvent(RedisCommandType command, RedisKey key) { + if (listeners.isEmpty()) { + return; + } + + List<EventListener> listenerList = listeners.get(key); + if (listenerList == null) { + return; + } + + for (EventListener listener : listenerList) { + if (listener.process(command, key) == EventResponse.REMOVE_AND_STOP) { + removeListener(listener); + break; + } + } + } + + /** + * The total number of keys registered by all listeners (includes duplicates). + */ + @VisibleForTesting + public int size() { + return keysRegistered; + } + + @Override + public synchronized void afterBucketRemoved(int bucketId, Iterable<?> keys) { + Set<EventListener> resubmittingList = new HashSet<>(); + for (Map.Entry<RedisKey, List<EventListener>> entry : listeners.entrySet()) { + if (entry.getKey().getBucketId() == bucketId) { + resubmittingList.addAll(entry.getValue()); + } + } + + resubmittingList.forEach(x -> { + x.resubmitCommand(); + removeListener(x); + }); + } + + private synchronized void removeListener(EventListener listener) { + boolean listenerRemoved = false; + for (RedisKey key : listener.keys()) { + List<EventListener> listenerList = listeners.get(key); + listenerRemoved |= listenerList.remove(listener); + if (listenerList.isEmpty()) { + listeners.remove(key); + } + } + + if (listenerRemoved) { + keysRegistered -= listener.keys().size(); + if (listener.getTimeout() > 0) { + timerTasks.remove(listener).cancel(); Review comment: This "cancel()" call could very likely lead to what looks like a memory leak. Timer does not actually remove the listener from its internal queue when you call cancel(). Instead it just marks it as cancelled. Then, if it ever get to the point in time when it would have expired, it will find it and see that it was cancelled. So if users have a long timeout, because the are very patient, then this will hold on to all those listeners and everything they reference for that long period of time even though we called cancel() quickly. You have two easy options: 1. keep using Timer but keep track of how many times you have called cancel and when that hits a certain number (how ever many zombie listeners you want to allow) call Timer.purge(). 2. use ScheduledThreadPoolExecutor instead of timer call call setRemoveOnCancelPolicy(true). This cause the remove to be done right when you call cancel. The remove has to search the queue for the listener so with a big queue this could be expensive. But this has the advantage of one thread not getting stuck with doing a possibly expensive purge call. ScheduledThreadPoolExecutor is more modern and has some possibly useful features that Timer does not. See the javadocs Timer class comment. If suggests using ScheduledThreadPoolExecutor instead of it. ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/EventDistributor.java ########## @@ -0,0 +1,128 @@ +/* + * 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.geode.redis.internal.eventing; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.Timer; +import java.util.TimerTask; + +import org.apache.logging.log4j.Logger; + +import org.apache.geode.annotations.VisibleForTesting; +import org.apache.geode.cache.partition.PartitionListenerAdapter; +import org.apache.geode.logging.internal.log4j.api.LogService; +import org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; + +public class EventDistributor extends PartitionListenerAdapter { + + private static final Logger logger = LogService.getLogger(); + private final Map<RedisKey, List<EventListener>> listeners = new HashMap<>(); + private final Map<EventListener, TimerTask> timerTasks = new HashMap<>(); Review comment: Another option to consider is to expand EventListener to allow its TimerTask to be added to it. Then you just have one map "listeners" instead of two. ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/netty/ExecutionHandlerContext.java ########## @@ -132,15 +135,17 @@ public ExecutionHandlerContext(Channel channel, this.serverPort = serverPort; this.member = member; this.securityService = securityService; + this.eventDistributor = eventDistributor; scanCursor = new BigInteger("0"); - channelId = channel.id(); redisStats.addClient(); channel.closeFuture().addListener(future -> logout()); } - public ChannelFuture writeToChannel(RedisResponse response) { - return client.writeToChannel(response); + public void writeToChannel(RedisResponse response) { + if (response != null) { Review comment: Would it be better to have a special RedisResponse.BLOCKED instead of using null? ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/EventListener.java ########## @@ -0,0 +1,58 @@ +/* + * 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.geode.redis.internal.eventing; + +import java.util.List; + +import org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; + +/** + * Interface intended to be implemented in order to receive Redis events. Specifically this would be + * keyspace events or blocking commands. EventListeners are registered with the + * {@link EventDistributor}. + */ +public interface EventListener { + + /** + * Receive and process an event. This method should execute very quickly. The return value + * determines additional process steps for the given event. + * + * @param commandType the command triggering the event + * @param key the key triggering the event + * @return response determining subsequent processing steps + */ + EventResponse process(RedisCommandType commandType, RedisKey key); + + /** + * Return the list of keys this listener is interested in. + */ + List<RedisKey> keys(); + + /** + * Method to resubmit a command if appropriate. This is only relevant for listeners that process + * events for blocking commands. Listeners that handle keyspace event notification will not use + * this. + */ + default void resubmitCommand() {}; + + /** + * Retrieve the timeout for this listener. The default is no timeout. Review comment: describe the time unitss ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/BlockingCommandListener.java ########## @@ -0,0 +1,99 @@ +/* + * 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.geode.redis.internal.eventing; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.apache.geode.redis.internal.commands.Command; +import org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; +import org.apache.geode.redis.internal.netty.Coder; +import org.apache.geode.redis.internal.netty.ExecutionHandlerContext; + +public class BlockingCommandListener implements EventListener { + + private final ExecutionHandlerContext context; + private final RedisCommandType command; + private final List<RedisKey> keys; + private final byte[][] commandOptions; + private final long timeout; + + /** + * Constructor to create an instance of a BlockingCommandListener in response to a blocking + * command. When receiving a relevant event, blocking commands simply resubmit the command + * into the Netty pipeline. + * + * @param context the associated ExecutionHandlerContext + * @param command the blocking command + * @param timeout the timeout for the command to block + * @param keys the list of keys the command is interested in + * @param commandOptions any additional options (other than the keys and timeout) required to + * resubmit the command. + */ + public BlockingCommandListener(ExecutionHandlerContext context, RedisCommandType command, + long timeout, List<RedisKey> keys, byte[]... commandOptions) { + this.context = context; + this.command = command; + this.timeout = timeout; + this.keys = Collections.unmodifiableList(keys); + this.commandOptions = commandOptions; + } + + @Override + public List<RedisKey> keys() { + return keys; + } + + @Override + public EventResponse process(RedisCommandType commandType, RedisKey key) { + if (!keys.contains(key)) { + return EventResponse.CONTIINUE; Review comment: typo "II" ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/EventDistributor.java ########## @@ -0,0 +1,128 @@ +/* + * 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.geode.redis.internal.eventing; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.Timer; +import java.util.TimerTask; + +import org.apache.logging.log4j.Logger; + +import org.apache.geode.annotations.VisibleForTesting; +import org.apache.geode.cache.partition.PartitionListenerAdapter; +import org.apache.geode.logging.internal.log4j.api.LogService; +import org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; + +public class EventDistributor extends PartitionListenerAdapter { + + private static final Logger logger = LogService.getLogger(); + private final Map<RedisKey, List<EventListener>> listeners = new HashMap<>(); + private final Map<EventListener, TimerTask> timerTasks = new HashMap<>(); + private final Timer timer = new Timer("GeodeForRedisEventTimer", true); + private int keysRegistered = 0; + + public synchronized void registerListener(EventListener listener) { + for (RedisKey key : listener.keys()) { + listeners.computeIfAbsent(key, k -> new ArrayList<>()).add(listener); + } + + if (listener.getTimeout() != 0) { + scheduleTimeout(listener, listener.getTimeout()); + } + + keysRegistered += listener.keys().size(); + } + + public synchronized void fireEvent(RedisCommandType command, RedisKey key) { + if (listeners.isEmpty()) { + return; + } + + List<EventListener> listenerList = listeners.get(key); + if (listenerList == null) { + return; + } + + for (EventListener listener : listenerList) { + if (listener.process(command, key) == EventResponse.REMOVE_AND_STOP) { + removeListener(listener); + break; + } + } + } + + /** + * The total number of keys registered by all listeners (includes duplicates). + */ + @VisibleForTesting + public int size() { + return keysRegistered; + } + + @Override + public synchronized void afterBucketRemoved(int bucketId, Iterable<?> keys) { + Set<EventListener> resubmittingList = new HashSet<>(); + for (Map.Entry<RedisKey, List<EventListener>> entry : listeners.entrySet()) { + if (entry.getKey().getBucketId() == bucketId) { + resubmittingList.addAll(entry.getValue()); + } + } + + resubmittingList.forEach(x -> { Review comment: This could take a while (i.e. processing the resubmittingList) since it has to send an error response to each client with a blocked command. So should it be done in a background thread instead of holding up whatever thread called afterBucketMoved? It definitely seems wrong to do this part while synchronized on "this". ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/EventDistributor.java ########## @@ -0,0 +1,128 @@ +/* + * 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.geode.redis.internal.eventing; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.Timer; +import java.util.TimerTask; + +import org.apache.logging.log4j.Logger; + +import org.apache.geode.annotations.VisibleForTesting; +import org.apache.geode.cache.partition.PartitionListenerAdapter; +import org.apache.geode.logging.internal.log4j.api.LogService; +import org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; + +public class EventDistributor extends PartitionListenerAdapter { + + private static final Logger logger = LogService.getLogger(); + private final Map<RedisKey, List<EventListener>> listeners = new HashMap<>(); + private final Map<EventListener, TimerTask> timerTasks = new HashMap<>(); + private final Timer timer = new Timer("GeodeForRedisEventTimer", true); + private int keysRegistered = 0; + + public synchronized void registerListener(EventListener listener) { + for (RedisKey key : listener.keys()) { + listeners.computeIfAbsent(key, k -> new ArrayList<>()).add(listener); + } + + if (listener.getTimeout() != 0) { + scheduleTimeout(listener, listener.getTimeout()); + } + + keysRegistered += listener.keys().size(); + } + + public synchronized void fireEvent(RedisCommandType command, RedisKey key) { + if (listeners.isEmpty()) { Review comment: if listeners is empty will calling listeners.get(key) be that expensive? It seems like you could probably get rid of the isEmpty check here ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/EventDistributor.java ########## @@ -0,0 +1,128 @@ +/* + * 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.geode.redis.internal.eventing; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.Timer; +import java.util.TimerTask; + +import org.apache.logging.log4j.Logger; + +import org.apache.geode.annotations.VisibleForTesting; +import org.apache.geode.cache.partition.PartitionListenerAdapter; +import org.apache.geode.logging.internal.log4j.api.LogService; +import org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; + +public class EventDistributor extends PartitionListenerAdapter { + + private static final Logger logger = LogService.getLogger(); + private final Map<RedisKey, List<EventListener>> listeners = new HashMap<>(); + private final Map<EventListener, TimerTask> timerTasks = new HashMap<>(); + private final Timer timer = new Timer("GeodeForRedisEventTimer", true); + private int keysRegistered = 0; + + public synchronized void registerListener(EventListener listener) { + for (RedisKey key : listener.keys()) { + listeners.computeIfAbsent(key, k -> new ArrayList<>()).add(listener); + } + + if (listener.getTimeout() != 0) { + scheduleTimeout(listener, listener.getTimeout()); + } + + keysRegistered += listener.keys().size(); + } + + public synchronized void fireEvent(RedisCommandType command, RedisKey key) { + if (listeners.isEmpty()) { + return; + } + + List<EventListener> listenerList = listeners.get(key); + if (listenerList == null) { + return; + } + + for (EventListener listener : listenerList) { + if (listener.process(command, key) == EventResponse.REMOVE_AND_STOP) { + removeListener(listener); + break; + } + } + } + + /** + * The total number of keys registered by all listeners (includes duplicates). + */ + @VisibleForTesting + public int size() { + return keysRegistered; + } + + @Override + public synchronized void afterBucketRemoved(int bucketId, Iterable<?> keys) { + Set<EventListener> resubmittingList = new HashSet<>(); + for (Map.Entry<RedisKey, List<EventListener>> entry : listeners.entrySet()) { + if (entry.getKey().getBucketId() == bucketId) { + resubmittingList.addAll(entry.getValue()); + } + } + + resubmittingList.forEach(x -> { + x.resubmitCommand(); + removeListener(x); + }); + } + + private synchronized void removeListener(EventListener listener) { + boolean listenerRemoved = false; + for (RedisKey key : listener.keys()) { + List<EventListener> listenerList = listeners.get(key); + listenerRemoved |= listenerList.remove(listener); + if (listenerList.isEmpty()) { + listeners.remove(key); + } + } + + if (listenerRemoved) { + keysRegistered -= listener.keys().size(); + if (listener.getTimeout() > 0) { + timerTasks.remove(listener).cancel(); + } + } + } + + private void scheduleTimeout(EventListener listener, long timeout) { + TimerTask task = new TimerTask() { Review comment: Another nice thing about using ScheduledThreadPoolExecutor is you can just have the EventListener implement Runnable. You will need to store the Future returned from calling submit somewhere so that you can call cancel on that Future if the timeout is hit. ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/EventDistributor.java ########## @@ -0,0 +1,128 @@ +/* + * 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.geode.redis.internal.eventing; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.Timer; +import java.util.TimerTask; + +import org.apache.logging.log4j.Logger; + +import org.apache.geode.annotations.VisibleForTesting; +import org.apache.geode.cache.partition.PartitionListenerAdapter; +import org.apache.geode.logging.internal.log4j.api.LogService; +import org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; + +public class EventDistributor extends PartitionListenerAdapter { + + private static final Logger logger = LogService.getLogger(); + private final Map<RedisKey, List<EventListener>> listeners = new HashMap<>(); + private final Map<EventListener, TimerTask> timerTasks = new HashMap<>(); + private final Timer timer = new Timer("GeodeForRedisEventTimer", true); + private int keysRegistered = 0; + + public synchronized void registerListener(EventListener listener) { + for (RedisKey key : listener.keys()) { + listeners.computeIfAbsent(key, k -> new ArrayList<>()).add(listener); + } + + if (listener.getTimeout() != 0) { + scheduleTimeout(listener, listener.getTimeout()); + } + + keysRegistered += listener.keys().size(); + } + + public synchronized void fireEvent(RedisCommandType command, RedisKey key) { Review comment: Having this method synchronized seems like it could hurt performance of conncurrent ops on different keys. They would all fight over this lock. It seems like your top level "listeners" map could be a ConcurrentHashMap and the nested ArrayList could be one of the Concurrent queue classes. -- 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: notifications-unsubscr...@geode.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
