ExpirationListener implements RedisFacade.RedisPubSub { @Override public void onPMessage(byte[] pattern, byte[] channelBuf, byte[] message) { if (channelBuf == null || message == null) { return; } if (!isExpiredChannel(channelBuf)) { return; } if (!isExpireKey(message)) { return; } String body = encode(message); if (logger.isDebugEnabled()) { logger.debug("Got notification for channel: '{}', body: '{}'", encode(channelBuf), body); } String sessionId = RedisSessionRepository.extractSessionId(body); logger.info("Session expired event for sessionId: '{}'", sessionId); sessionManager.deleteAsync(sessionId, true); } ExpirationListener(SessionManager sessionManager, String keyPrefix); @Override void onPMessage(byte[] pattern, byte[] channelBuf, byte[] message); @Override Object getLinked(); @Override void link(Object linkedImplementation);  }
@Test public void testOnPMessage() { String sessionId = "test-id"; byte[] pattern = SafeEncoder.encode(""); byte[] goodChannel = SafeEncoder.encode("_keyspace:test:expired"); byte[] notExpireChannel = SafeEncoder.encode("_keyspace:test:expared"); byte[] shortChannel = SafeEncoder.encode("expir"); byte[] goodKey = SafeEncoder.encode("myprefix:key:test-id"); byte[] shortKey = SafeEncoder.encode("myprefi"); expirationListener.onPMessage(pattern, null, goodKey); verify(sessionManager, never()).deleteAsync(sessionId, true); expirationListener.onPMessage(pattern, goodChannel, null); verify(sessionManager, never()).deleteAsync(sessionId, true); expirationListener.onPMessage(pattern, notExpireChannel, goodKey); verify(sessionManager, never()).deleteAsync(sessionId, true); expirationListener.onPMessage(pattern, shortChannel, goodKey); verify(sessionManager, never()).deleteAsync(sessionId, true); expirationListener.onPMessage(pattern, goodChannel, shortKey); verify(sessionManager, never()).deleteAsync(sessionId, true); expirationListener.onPMessage(pattern, goodChannel, goodKey); verify(sessionManager).deleteAsync(sessionId, true); }