HyperLogLog implemented. #18

pull/25/head
Nikita 11 years ago
parent 7580627630
commit 3aa32efa49

@ -68,7 +68,10 @@ public enum CommandType {
// Bits
BITCOUNT, BITOP, GETBIT, SETBIT;
BITCOUNT, BITOP, GETBIT, SETBIT,
// HyperLogLog
PFADD, PFCOUNT, PFMERGE;
public byte[] bytes;

@ -22,6 +22,7 @@ import org.redisson.connection.ConnectionManager;
import org.redisson.core.RAtomicLong;
import org.redisson.core.RCountDownLatch;
import org.redisson.core.RDeque;
import org.redisson.core.RHyperLogLog;
import org.redisson.core.RList;
import org.redisson.core.RLock;
import org.redisson.core.RMap;
@ -65,6 +66,7 @@ public class Redisson {
private final ConcurrentMap<String, RedissonSet> setsMap = new ReferenceMap<String, RedissonSet>(ReferenceType.STRONG, ReferenceType.SOFT);
private final ConcurrentMap<String, RedissonSortedSet> sortedSetMap = new ReferenceMap<String, RedissonSortedSet>(ReferenceType.STRONG, ReferenceType.SOFT);
private final ConcurrentMap<String, RedissonList> listsMap = new ReferenceMap<String, RedissonList>(ReferenceType.STRONG, ReferenceType.SOFT);
private final ConcurrentMap<String, RedissonHyperLogLog> hyperLogLogMap = new ReferenceMap<String, RedissonHyperLogLog>(ReferenceType.STRONG, ReferenceType.SOFT);
private final ConcurrentMap<String, RedissonMap> mapsMap = new ReferenceMap<String, RedissonMap>(ReferenceType.STRONG, ReferenceType.SOFT);
private final ConnectionManager connectionManager;
@ -99,6 +101,19 @@ public class Redisson {
return new Redisson(config);
}
public <V> RHyperLogLog<V> getHyperLogLog(String name) {
RedissonHyperLogLog<V> logLog = hyperLogLogMap.get(name);
if (logLog == null) {
logLog = new RedissonHyperLogLog<V>(connectionManager, name);
RedissonHyperLogLog<V> oldLogLog = hyperLogLogMap.putIfAbsent(name, logLog);
if (oldLogLog != null) {
logLog = oldLogLog;
}
}
return logLog;
}
/**
* Returns distributed list instance by name.
*

@ -0,0 +1,115 @@
/**
* Copyright 2014 Nikita Koksharov, Nickolay Borbit
*
* Licensed 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.redisson;
import java.util.Collection;
import java.util.concurrent.Future;
import org.redisson.connection.ConnectionManager;
import org.redisson.core.RHyperLogLog;
import com.lambdaworks.redis.RedisConnection;
public class RedissonHyperLogLog<V> extends RedissonObject implements RHyperLogLog<V> {
private final ConnectionManager connectionManager;
RedissonHyperLogLog(ConnectionManager connectionManager, String name) {
super(name);
this.connectionManager = connectionManager;
}
@Override
public long add(V obj) {
RedisConnection<String, Object> conn = connectionManager.connectionWriteOp();
try {
return conn.pfadd(getName(), obj);
} finally {
connectionManager.release(conn);
}
}
@Override
public long addAll(Collection<V> objects) {
RedisConnection<String, Object> conn = connectionManager.connectionWriteOp();
try {
return conn.pfadd(getName(), objects.toArray());
} finally {
connectionManager.release(conn);
}
}
@Override
public long count() {
RedisConnection<String, Object> conn = connectionManager.connectionReadOp();
try {
return conn.pfcount(getName());
} finally {
connectionManager.release(conn);
}
}
@Override
public long countWith(String... otherLogNames) {
RedisConnection<String, Object> conn = connectionManager.connectionReadOp();
try {
return conn.pfcount(getName(), otherLogNames);
} finally {
connectionManager.release(conn);
}
}
@Override
public long mergeWith(String... otherLogNames) {
RedisConnection<String, Object> conn = connectionManager.connectionReadOp();
try {
return conn.pfmerge(getName(), otherLogNames);
} finally {
connectionManager.release(conn);
}
}
@Override
public Future<Long> addAsync(V obj) {
RedisConnection<String, Object> conn = connectionManager.connectionWriteOp();
return conn.getAsync().pfadd(getName(), obj).addListener(connectionManager.createListener(conn));
}
@Override
public Future<Long> addAllAsync(Collection<V> objects) {
RedisConnection<String, Object> conn = connectionManager.connectionWriteOp();
return conn.getAsync().pfadd(getName(), objects.toArray()).addListener(connectionManager.createListener(conn));
}
@Override
public Future<Long> countAsync() {
RedisConnection<String, Object> conn = connectionManager.connectionReadOp();
return conn.getAsync().pfcount(getName()).addListener(connectionManager.createListener(conn));
}
@Override
public Future<Long> countWithAsync(String... otherLogNames) {
RedisConnection<String, Object> conn = connectionManager.connectionReadOp();
return conn.getAsync().pfcount(getName(), otherLogNames).addListener(connectionManager.createListener(conn));
}
@Override
public Future<Long> mergeWithAsync(String... otherLogNames) {
RedisConnection<String, Object> conn = connectionManager.connectionWriteOp();
return conn.getAsync().pfmerge(getName(), otherLogNames).addListener(connectionManager.createListener(conn));
}
}

@ -0,0 +1,43 @@
/**
* Copyright 2014 Nikita Koksharov, Nickolay Borbit
*
* Licensed 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.redisson.core;
import java.util.Collection;
import java.util.concurrent.Future;
public interface RHyperLogLog<V> extends RObject {
long add(V obj);
long addAll(Collection<V> objects);
long count();
long countWith(String ... otherLogNames);
long mergeWith(String ... otherLogNames);
Future<Long> addAsync(V obj);
Future<Long> addAllAsync(Collection<V> objects);
Future<Long> countAsync();
Future<Long> countWithAsync(String ... otherLogNames);
Future<Long> mergeWithAsync(String ... otherLogNames);
}

@ -0,0 +1,19 @@
package org.redisson;
import org.junit.Assert;
import org.junit.Test;
import org.redisson.core.RHyperLogLog;
public class RedissonHyperLogLogTest extends BaseTest {
@Test
public void testAdd() {
RHyperLogLog<Integer> log = redisson.getHyperLogLog("log");
log.add(1);
log.add(2);
log.add(3);
Assert.assertEquals(3L, log.count());
}
}
Loading…
Cancel
Save