JedisClusterFacade extends AbstractJedisFacade { @Override public void psubscribe(final RedisPubSub listener, String pattern) { BinaryJedisPubSub bps = new BinaryJedisPubSub() { @Override public void onPMessage(byte[] pattern, byte[] channel, byte[] message) { listener.onPMessage(pattern, channel, message); } }; listener.link(bps); jedisCluster.psubscribe(bps, encode(pattern)); } JedisClusterFacade(TransactionalJedisCluster jedisCluster); @Override void psubscribe(final RedisPubSub listener, String pattern); @Override void punsubscribe(final RedisPubSub listener, byte[] pattern); @Override Long hdel(byte[] key, byte[]... fields); @Override List<byte[]> hmget(byte[] key, byte[]... fields); @Override String hmset(byte[] key, Map<byte[], byte[]> hash); @Override Long hsetnx(final byte[] key, final byte[] field, final byte[] value); @Override Long hset(final byte[] key, final byte[] field, final byte[] value); @Override Set<byte[]> hkeys(byte[] key); @Override String set(byte[] key, byte[] value); @Override String setex(byte[] key, int expiry, byte[] value); @Override Long expire(byte[] key, int value); @Override void srem(byte[] key, byte[]... member); @Override Long sadd(byte[] key, byte[]... member); @Override Long del(byte[]... keys); @Override Boolean exists(byte[] key); @Override Set<byte[]> smembers(byte[] key); @Override Set<byte[]> spop(byte[] key, long count); @Override Long expireAt(byte[] key, long unixTime); @Override Long zadd(byte[] key, double score, byte[] elem); @Override Long zrem(byte[] key, byte[]... fields); @Override Set<byte[]> zrangeByScore(byte[] key, double start, double end); @Override Set<byte[]> zrange(byte[] key, long start, long end); @Override Long persist(byte[] key); @Override String info(String section); @Override void close(); @Override void requestFinished(); @Override String rename(byte[] oldkey, byte[] newkey); @Override byte[] get(byte[] key); @Override Long publish(byte[] channel, byte[] message); @Override void startMonitoring(MetricRegistry metrics); @Override RedisFacade.ResponseFacade<T> transaction(final byte[] key, final TransactionRunner<T> transaction); void setTransactionOnKey(boolean transactionOnKey);  }
@Test public void testPsubscribe() { BinaryJedisPubSub listener = mock(BinaryJedisPubSub.class); RedisFacade.RedisPubSub facadeListener = mock(RedisFacade.RedisPubSub.class); when(facadeListener.getLinked()).thenReturn(listener); String pattern = "test"; rf.psubscribe(facadeListener, pattern); ArgumentCaptor<BinaryJedisPubSub> capture = ArgumentCaptor.forClass(BinaryJedisPubSub.class); verify(facadeListener).link(capture.capture()); verify(jedisCluster).psubscribe(capture.getValue(), SafeEncoder.encode(pattern)); }