Implement pfmerge command the correct way #158

The pfmerge command invoked PFADD instead of PFMERGE including a wrong method signature. This commit changes the signature from Long pfmerge(...) to String pfmerge() and respective for the async API and the pfmerge method does no longer invoke the PFADD command but the PFMERGE command.

src/main/java/com/lambdaworks/redis/RedisAsyncConnectionImpl.java

@@ -1511,7 +1511,7 @@ public RedisFuture<Long> pfadd(K key, V value, V... moreValues) {
     }
 
     @Override
-    public RedisFuture<Long> pfmerge(K destkey, K sourcekey, K... moreSourceKeys) {
+    public RedisFuture<String> pfmerge(K destkey, K sourcekey, K... moreSourceKeys) {
         return dispatch(commandBuilder.pfmerge(destkey, sourcekey, moreSourceKeys));
     }
 

src/main/java/com/lambdaworks/redis/RedisCommandBuilder.java

@@ -1571,14 +1571,14 @@ public Command<K, V, Long> pfcount(K key, K... moreKeys) {
     }
 
     @SuppressWarnings("unchecked")
-    public Command<K, V, Long> pfmerge(K destkey, K sourcekey, K... moreSourceKeys) {
+    public Command<K, V, String> pfmerge(K destkey, K sourcekey, K... moreSourceKeys) {
         assertNotNull(destkey, "destkey " + MUST_NOT_BE_NULL);
         assertNotNull(sourcekey, "sourcekey " + MUST_NOT_BE_NULL);
         assertNotNull(moreSourceKeys, "moreSourceKeys " + MUST_NOT_BE_NULL);
         assertNoNullElements(moreSourceKeys, "moreSourceKeys " + MUST_NOT_CONTAIN_NULL_ELEMENTS);
 
         CommandArgs<K, V> args = new CommandArgs<K, V>(codec).addKeys(destkey).addKey(sourcekey).addKeys(moreSourceKeys);
-        return createCommand(PFADD, new IntegerOutput<K, V>(codec), args);
+        return createCommand(PFMERGE, new StatusOutput<K, V>(codec), args);
     }
 
     public Command<K, V, String> clusterMeet(String ip, int port) {

src/main/java/com/lambdaworks/redis/RedisHLLAsyncConnection.java

@@ -29,9 +29,9 @@ public interface RedisHLLAsyncConnection<K, V> {
      * @param sourcekey the source key
      * @param moreSourceKeys more source keys
      * 
-     * @return RedisFuture&lt;Long&gt; simple-string-reply The command just returns {@code OK}.
+     * @return RedisFuture&lt;String&gt; simple-string-reply The command just returns {@code OK}.
      */
-    RedisFuture<Long> pfmerge(K destkey, K sourcekey, K... moreSourceKeys);
+    RedisFuture<String> pfmerge(K destkey, K sourcekey, K... moreSourceKeys);
 
     /**
      * Return the approximated cardinality of the set(s) observed by the HyperLogLog at key(s).

src/main/java/com/lambdaworks/redis/RedisHLLConnection.java

@@ -31,7 +31,7 @@ public interface RedisHLLConnection<K, V> {
      * 
      * @return Long simple-string-reply The command just returns {@code OK}.
      */
-    Long pfmerge(K destkey, K sourcekey, K... moreSourceKeys);
+    String pfmerge(K destkey, K sourcekey, K... moreSourceKeys);
 
     /**
      * Return the approximated cardinality of the set(s) observed by the HyperLogLog at key(s).

src/test/java/com/lambdaworks/redis/HLLCommandTest.java

@@ -5,6 +5,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
 
+import com.lambdaworks.redis.cluster.SlotHash;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
@@ -42,7 +43,15 @@ public void pfmerge() throws Exception {
         redis.pfadd("key2", "value2");
         redis.pfadd("key3", "value3");
 
-        assertThat(redis.pfmerge(key, "key2", "key3")).isEqualTo(1);
+        assertThat(redis.pfmerge(key, "key2", "key3")).isEqualTo("OK");
+        assertThat(redis.pfcount(key)).isEqualTo(3);
+
+        redis.pfadd("key2660", "rand", "mat");
+        redis.pfadd("key7112", "mat", "perrin");
+
+        redis.pfmerge("key8885", "key2660", "key7112");
+
+        assertThat(redis.pfcount("key8885")).isEqualTo(3);
     }
 
     @Test

src/test/java/com/lambdaworks/redis/cluster/RedisClusterClientTest.java

@@ -5,7 +5,9 @@
 import static com.lambdaworks.redis.cluster.ClusterTestUtil.getNodeId;
 import static com.lambdaworks.redis.cluster.ClusterTestUtil.getOwnPartition;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.fail;
+import static org.junit.Assume.assumeThat;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -672,4 +674,20 @@ public void testReadFromNull() throws Exception {
 
         connection.close();
     }
+
+    @Test
+    public void testPfmerge() throws Exception {
+        RedisAdvancedClusterConnection<String, String> connection = clusterClient.connectCluster();
+
+        assertThat(SlotHash.getSlot("key2660")).isEqualTo(SlotHash.getSlot("key7112")).isEqualTo(SlotHash.getSlot("key8885"));
+
+        connection.pfadd("key2660", "rand", "mat");
+        connection.pfadd("key7112", "mat", "perrin");
+
+        connection.pfmerge("key8885", "key2660", "key7112");
+
+        assertThat(connection.pfcount("key8885")).isEqualTo(3);
+
+        connection.close();
+    }
 }