DonalEvans commented on a change in pull request #7408: URL: https://github.com/apache/geode/pull/7408#discussion_r823005434
########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/commands/executor/list/BLPopExecutor.java ########## @@ -0,0 +1,52 @@ +/* + * 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.commands.executor.list; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.geode.redis.internal.RedisConstants; +import org.apache.geode.redis.internal.commands.Command; +import org.apache.geode.redis.internal.commands.executor.CommandExecutor; +import org.apache.geode.redis.internal.commands.executor.RedisResponse; +import org.apache.geode.redis.internal.data.RedisKey; +import org.apache.geode.redis.internal.data.RedisList; +import org.apache.geode.redis.internal.netty.Coder; +import org.apache.geode.redis.internal.netty.ExecutionHandlerContext; + +public class BLPopExecutor implements CommandExecutor { + + @Override + public RedisResponse executeCommand(Command command, ExecutionHandlerContext context) { + List<byte[]> arguments = command.getCommandArguments(); + int keyCount = arguments.size() - 1; Review comment: This variable name is confusing given how it's used when determining the timeout. Should it be something like `timeoutIndex` instead? ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/commands/executor/list/BLPopExecutor.java ########## @@ -0,0 +1,52 @@ +/* + * 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.commands.executor.list; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.geode.redis.internal.RedisConstants; +import org.apache.geode.redis.internal.commands.Command; +import org.apache.geode.redis.internal.commands.executor.CommandExecutor; +import org.apache.geode.redis.internal.commands.executor.RedisResponse; +import org.apache.geode.redis.internal.data.RedisKey; +import org.apache.geode.redis.internal.data.RedisList; +import org.apache.geode.redis.internal.netty.Coder; +import org.apache.geode.redis.internal.netty.ExecutionHandlerContext; + +public class BLPopExecutor implements CommandExecutor { + + @Override + public RedisResponse executeCommand(Command command, ExecutionHandlerContext context) { + List<byte[]> arguments = command.getCommandArguments(); + int keyCount = arguments.size() - 1; + double timeoutMillis; + try { + timeoutMillis = Coder.bytesToDouble(arguments.get(keyCount)) * 1000; + } catch (NumberFormatException e) { + return RedisResponse.error(RedisConstants.ERROR_TIMEOUT_INVALID); + } Review comment: We also need a check that the timeout isn't negative. If it is, Redis returns `ERR timeout is negative`. ########## File path: geode-for-redis/src/integrationTest/java/org/apache/geode/redis/internal/commands/executor/list/AbstractBLPopIntegrationTest.java ########## @@ -0,0 +1,142 @@ +/* + * 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.commands.executor.list; + +import static org.apache.geode.test.dunit.rules.RedisClusterStartupRule.BIND_ADDRESS; +import static org.apache.geode.test.dunit.rules.RedisClusterStartupRule.REDIS_CLIENT_TIMEOUT; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.After; +import org.junit.Before; +import org.junit.ClassRule; +import org.junit.Test; +import redis.clients.jedis.HostAndPort; +import redis.clients.jedis.JedisCluster; +import redis.clients.jedis.Protocol; + +import org.apache.geode.redis.RedisIntegrationTest; +import org.apache.geode.redis.internal.RedisConstants; +import org.apache.geode.test.awaitility.GeodeAwaitility; +import org.apache.geode.test.junit.rules.ExecutorServiceRule; + +public abstract class AbstractBLPopIntegrationTest implements RedisIntegrationTest { + private static final String KEY = "key"; + + protected JedisCluster jedis; + + public abstract void awaitEventDistributorSize(int size) throws Exception; + + @ClassRule + public static ExecutorServiceRule executor = new ExecutorServiceRule(); + + @Before + public void setUp() { + jedis = new JedisCluster(new HostAndPort(BIND_ADDRESS, getPort()), REDIS_CLIENT_TIMEOUT); + } + + @After + public void tearDown() { + flushAll(); + jedis.close(); + } + + @Test + public void testInvalidArguments_throwErrors() { + assertThatThrownBy(() -> jedis.sendCommand("key", Protocol.Command.BLPOP)) + .hasMessageContaining("ERR wrong number of arguments for 'blpop' command"); + assertThatThrownBy(() -> jedis.sendCommand("key1", Protocol.Command.BLPOP, "key")) + .hasMessageContaining("ERR wrong number of arguments for 'blpop' command"); + } + + @Test + public void testInvalidTimeout_throwsError() { + assertThatThrownBy(() -> jedis.sendCommand("key1", Protocol.Command.BLPOP, "key1", + "0.A")) + .hasMessageContaining(RedisConstants.ERROR_TIMEOUT_INVALID); Review comment: Could the `hasMessageContaining()` calls in this class be replaced with `hasMessage()` to be consistent with the changes from GEODE-10049? ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/commands/executor/list/BLPopExecutor.java ########## @@ -0,0 +1,52 @@ +/* + * 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.commands.executor.list; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.geode.redis.internal.RedisConstants; +import org.apache.geode.redis.internal.commands.Command; +import org.apache.geode.redis.internal.commands.executor.CommandExecutor; +import org.apache.geode.redis.internal.commands.executor.RedisResponse; +import org.apache.geode.redis.internal.data.RedisKey; +import org.apache.geode.redis.internal.data.RedisList; +import org.apache.geode.redis.internal.netty.Coder; +import org.apache.geode.redis.internal.netty.ExecutionHandlerContext; + +public class BLPopExecutor implements CommandExecutor { + + @Override + public RedisResponse executeCommand(Command command, ExecutionHandlerContext context) { + List<byte[]> arguments = command.getCommandArguments(); + int keyCount = arguments.size() - 1; + double timeoutMillis; + try { + timeoutMillis = Coder.bytesToDouble(arguments.get(keyCount)) * 1000; + } catch (NumberFormatException e) { + return RedisResponse.error(RedisConstants.ERROR_TIMEOUT_INVALID); + } + + List<RedisKey> keys = new ArrayList<>(keyCount); + for (int i = 0; i < keyCount; i++) { + keys.add(new RedisKey(arguments.get(i))); + } + + List<byte[]> popped = context.lockedExecute(keys.get(0), keys, + () -> RedisList.blpop(context, keys, (int) timeoutMillis)); Review comment: Inside `BlockingCommandListener.java`, the timeout is stored as a `long`. If we decide to use a double rather than an integer for the timeout argument, it might be best to have `timeoutMillis` be a `long` and assign it using: ``` timeoutMillis = (long) (bytesToDouble(arguments.get(keyCount)) * 1000); ``` or to change the timeout to be an `int` throughout. ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/EventResponse.java ########## @@ -0,0 +1,33 @@ +/* + * 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 org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; + +/** + * Response returned by {@link EventListener#process(RedisCommandType, RedisKey)} + */ +public enum EventResponse { + /** + * This response would be used by listeners handling keyspace events + */ + CONTINUE, + /** + * This response would be used by listeners handling blocking commands + */ + REMOVE_AND_STOP; Review comment: Unnecessary semicolon here. ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/commands/executor/list/BLPopExecutor.java ########## @@ -0,0 +1,52 @@ +/* + * 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.commands.executor.list; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.geode.redis.internal.RedisConstants; +import org.apache.geode.redis.internal.commands.Command; +import org.apache.geode.redis.internal.commands.executor.CommandExecutor; +import org.apache.geode.redis.internal.commands.executor.RedisResponse; +import org.apache.geode.redis.internal.data.RedisKey; +import org.apache.geode.redis.internal.data.RedisList; +import org.apache.geode.redis.internal.netty.Coder; +import org.apache.geode.redis.internal.netty.ExecutionHandlerContext; + +public class BLPopExecutor implements CommandExecutor { + + @Override + public RedisResponse executeCommand(Command command, ExecutionHandlerContext context) { + List<byte[]> arguments = command.getCommandArguments(); + int keyCount = arguments.size() - 1; + double timeoutMillis; + try { + timeoutMillis = Coder.bytesToDouble(arguments.get(keyCount)) * 1000; Review comment: Specifying the timeout as a double is a feature new in Redis 6. In other commands added for Lists (specifically LPOP and RPOP) we've matched the behaviour of Redis 5. For consistency, should the implementation of BLPOP be taking only integer values for timeout? ########## File path: geode-for-redis/src/integrationTest/java/org/apache/geode/redis/internal/commands/executor/list/AbstractBLPopIntegrationTest.java ########## @@ -0,0 +1,142 @@ +/* + * 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.commands.executor.list; + +import static org.apache.geode.test.dunit.rules.RedisClusterStartupRule.BIND_ADDRESS; +import static org.apache.geode.test.dunit.rules.RedisClusterStartupRule.REDIS_CLIENT_TIMEOUT; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.After; +import org.junit.Before; +import org.junit.ClassRule; +import org.junit.Test; +import redis.clients.jedis.HostAndPort; +import redis.clients.jedis.JedisCluster; +import redis.clients.jedis.Protocol; + +import org.apache.geode.redis.RedisIntegrationTest; +import org.apache.geode.redis.internal.RedisConstants; +import org.apache.geode.test.awaitility.GeodeAwaitility; +import org.apache.geode.test.junit.rules.ExecutorServiceRule; + +public abstract class AbstractBLPopIntegrationTest implements RedisIntegrationTest { + private static final String KEY = "key"; + + protected JedisCluster jedis; + + public abstract void awaitEventDistributorSize(int size) throws Exception; + + @ClassRule + public static ExecutorServiceRule executor = new ExecutorServiceRule(); + + @Before + public void setUp() { + jedis = new JedisCluster(new HostAndPort(BIND_ADDRESS, getPort()), REDIS_CLIENT_TIMEOUT); + } + + @After + public void tearDown() { + flushAll(); + jedis.close(); + } + + @Test + public void testInvalidArguments_throwErrors() { + assertThatThrownBy(() -> jedis.sendCommand("key", Protocol.Command.BLPOP)) + .hasMessageContaining("ERR wrong number of arguments for 'blpop' command"); + assertThatThrownBy(() -> jedis.sendCommand("key1", Protocol.Command.BLPOP, "key")) + .hasMessageContaining("ERR wrong number of arguments for 'blpop' command"); + } Review comment: This can be simplified to: ``` @Test public void testInvalidNumberOfArguments_Errors() { assertAtLeastNArgs(jedis, Protocol.Command.BLPOP, 2); } ``` ########## File path: geode-for-redis/src/integrationTest/java/org/apache/geode/redis/internal/commands/executor/list/AbstractBLPopIntegrationTest.java ########## @@ -0,0 +1,142 @@ +/* + * 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.commands.executor.list; + +import static org.apache.geode.test.dunit.rules.RedisClusterStartupRule.BIND_ADDRESS; +import static org.apache.geode.test.dunit.rules.RedisClusterStartupRule.REDIS_CLIENT_TIMEOUT; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.After; +import org.junit.Before; +import org.junit.ClassRule; +import org.junit.Test; +import redis.clients.jedis.HostAndPort; +import redis.clients.jedis.JedisCluster; +import redis.clients.jedis.Protocol; + +import org.apache.geode.redis.RedisIntegrationTest; +import org.apache.geode.redis.internal.RedisConstants; +import org.apache.geode.test.awaitility.GeodeAwaitility; +import org.apache.geode.test.junit.rules.ExecutorServiceRule; + +public abstract class AbstractBLPopIntegrationTest implements RedisIntegrationTest { + private static final String KEY = "key"; + + protected JedisCluster jedis; + + public abstract void awaitEventDistributorSize(int size) throws Exception; + + @ClassRule + public static ExecutorServiceRule executor = new ExecutorServiceRule(); + + @Before + public void setUp() { + jedis = new JedisCluster(new HostAndPort(BIND_ADDRESS, getPort()), REDIS_CLIENT_TIMEOUT); + } + + @After + public void tearDown() { + flushAll(); + jedis.close(); + } + + @Test + public void testInvalidArguments_throwErrors() { + assertThatThrownBy(() -> jedis.sendCommand("key", Protocol.Command.BLPOP)) + .hasMessageContaining("ERR wrong number of arguments for 'blpop' command"); + assertThatThrownBy(() -> jedis.sendCommand("key1", Protocol.Command.BLPOP, "key")) + .hasMessageContaining("ERR wrong number of arguments for 'blpop' command"); + } + + @Test + public void testInvalidTimeout_throwsError() { Review comment: We need to also add a test for when the timeout is negative. ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/EventListener.java ########## @@ -0,0 +1,68 @@ +/* + * 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: Should this comment specify the units for the timeout? ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/EventDistributor.java ########## @@ -0,0 +1,121 @@ +/* + * 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.HashSet; +import java.util.Map; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import org.apache.geode.annotations.VisibleForTesting; +import org.apache.geode.cache.partition.PartitionListenerAdapter; +import org.apache.geode.logging.internal.executors.LoggingThreadFactory; +import org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; + +public class EventDistributor extends PartitionListenerAdapter { + + private final Map<RedisKey, Queue<EventListener>> listeners = new ConcurrentHashMap<>(); + + private final ScheduledThreadPoolExecutor timerExecutor = + new ScheduledThreadPoolExecutor(1, + new LoggingThreadFactory("GeodeForRedisEventTimer-", true)); + + private int keysRegistered = 0; + + public EventDistributor() { + timerExecutor.setRemoveOnCancelPolicy(true); + } + + public synchronized void registerListener(EventListener listener) { + for (RedisKey key : listener.keys()) { + listeners.computeIfAbsent(key, k -> new LinkedBlockingQueue<>()).add(listener); + } + + if (listener.getTimeout() != 0) { + scheduleTimeout(listener, listener.getTimeout()); + } + + keysRegistered += listener.keys().size(); + } + + public void fireEvent(RedisCommandType command, RedisKey key) { + Queue<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 void afterBucketRemoved(int bucketId, Iterable<?> keys) { + Set<EventListener> resubmittingList = new HashSet<>(); + for (Map.Entry<RedisKey, Queue<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()) { + Queue<EventListener> listenerList = listeners.get(key); + if (listenerList == null) { + continue; + } + listenerRemoved |= listenerList.remove(listener); + if (listenerList.isEmpty()) { + listeners.remove(key); + } + } + + if (listenerRemoved) { + keysRegistered -= listener.keys().size(); + listener.cleanup(); + } + } + + private void scheduleTimeout(EventListener listener, long timeout) { + ScheduledFuture<?> scheduledTask = + timerExecutor.schedule(() -> removeListener(listener), timeout, TimeUnit.MILLISECONDS); + listener.setCleanupTask(() -> scheduledTask.cancel(false)); + } Review comment: If BLPOP is executed with a non-zero timeout, then once that timeout elapses, the client should receive a `null` response. The below test passes with open source Redis but fails with the current implementation in geode-for-redis: ``` @Test public void testBLPopWhenTimeoutIsExceeded() { int timeout = 10; Future<List<String>> future = executor.submit(() -> jedis.blpop(timeout, KEY)); await().atMost(timeout * 2, TimeUnit.SECONDS).untilAsserted(() -> assertThat(future.get()).isNull()); } ``` -- 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
