RedisIO { public static Write write() { return new AutoValue_RedisIO_Write.Builder() .setConnectionConfiguration(RedisConnectionConfiguration.create()) .setMethod(Write.Method.APPEND) .build(); } private  RedisIO(); static Read read(); @Deprecated static ReadAll readAll(); static ReadKeyPatterns readKeyPatterns(); static Write write();  }
@Test public void testWriteWithMethodSet() { String key = "testWriteWithMethodSet"; client.set(key, "value"); String newValue = "newValue"; PCollection<KV<String, String>> write = p.apply(Create.of(KV.of(key, newValue))); write.apply(RedisIO.write().withEndpoint(REDIS_HOST, port).withMethod(Method.SET)); p.run(); assertEquals(newValue, client.get(key)); }
@Test public void testWriteWithMethodLPush() { String key = "testWriteWithMethodLPush"; String value = "value"; client.lpush(key, value); String newValue = "newValue"; PCollection<KV<String, String>> write = p.apply(Create.of(KV.of(key, newValue))); write.apply(RedisIO.write().withEndpoint(REDIS_HOST, port).withMethod(Method.LPUSH)); p.run(); List<String> values = client.lrange(key, 0, -1); assertEquals(newValue + value, String.join("", values)); }
@Test public void testWriteWithMethodRPush() { String key = "testWriteWithMethodRPush"; String value = "value"; client.lpush(key, value); String newValue = "newValue"; PCollection<KV<String, String>> write = p.apply(Create.of(KV.of(key, newValue))); write.apply(RedisIO.write().withEndpoint(REDIS_HOST, port).withMethod(Method.RPUSH)); p.run(); List<String> values = client.lrange(key, 0, -1); assertEquals(value + newValue, String.join("", values)); }
@Test public void testWriteWithMethodSAdd() { String key = "testWriteWithMethodSAdd"; List<String> values = Arrays.asList("0", "1", "2", "3", "2", "4", "0", "5"); List<KV<String, String>> data = buildConstantKeyList(key, values); PCollection<KV<String, String>> write = p.apply(Create.of(data)); write.apply(RedisIO.write().withEndpoint(REDIS_HOST, port).withMethod(Method.SADD)); p.run(); Set<String> expected = new HashSet<>(values); Set<String> members = client.smembers(key); assertEquals(expected, members); }
@Test public void testWriteWithMethodPFAdd() { String key = "testWriteWithMethodPFAdd"; List<String> values = Arrays.asList("0", "1", "2", "3", "2", "4", "0", "5"); List<KV<String, String>> data = buildConstantKeyList(key, values); PCollection<KV<String, String>> write = p.apply(Create.of(data)); write.apply(RedisIO.write().withEndpoint(REDIS_HOST, port).withMethod(Method.PFADD)); p.run(); long count = client.pfcount(key); assertEquals(6, count); }
@Test public void testWriteUsingINCRBY() { String key = "key_incr"; List<String> values = Arrays.asList("0", "1", "2", "-3", "2", "4", "0", "5"); List<KV<String, String>> data = buildConstantKeyList(key, values); PCollection<KV<String, String>> write = p.apply(Create.of(data)); write.apply(RedisIO.write().withEndpoint(REDIS_HOST, port).withMethod(Method.INCRBY)); p.run(); long count = Long.parseLong(client.get(key)); assertEquals(11, count); }
@Test public void testWriteUsingDECRBY() { String key = "key_decr"; List<String> values = Arrays.asList("-10", "1", "2", "-3", "2", "4", "0", "5"); List<KV<String, String>> data = buildConstantKeyList(key, values); PCollection<KV<String, String>> write = p.apply(Create.of(data)); write.apply(RedisIO.write().withEndpoint(REDIS_HOST, port).withMethod(Method.DECRBY)); p.run(); long count = Long.parseLong(client.get(key)); assertEquals(-1, count); }