// Path: src/main/java/org/rakam/cache/hazelcast/hyperloglog/HyperLogLogService.java
// public class HyperLogLogService implements ManagedService, RemoteService, MigrationAwareService {
// 
//     public static final String SERVICE_NAME = "rakam:hyperLogLogService";
// 
//     private NodeEngine nodeEngine;
//     private final ConcurrentMap<String, HLLWrapper> containers = new ConcurrentHashMap<String, HLLWrapper>();
//     private final ConstructorFunction<String, HLLWrapper> CountersConstructorFunction =
//             new ConstructorFunction<String, HLLWrapper>() {
//                 public HLLWrapper createNew(String key) {
//                     return new HLLWrapper();
//                 }
//             };
// 
//     public HyperLogLogService() {
//     }
// 
//     public HLLWrapper getHLL(String name) {
//         return getOrPutIfAbsent(containers, name, CountersConstructorFunction);
//     }
// 
//     @Override
//     public void init(NodeEngine nodeEngine, Properties properties) {
//         this.nodeEngine = nodeEngine;
//     }
// 
//     @Override
//     public void reset() {
//         containers.clear();
//     }
// 
//     @Override
//     public void shutdown(boolean terminate) {
//         reset();
//     }
// 
//     @Override
//     public HyperLogLogProxy createDistributedObject(String name) {
//         return new HyperLogLogProxy(name, nodeEngine, this);
//     }
// 
//     @Override
//     public void destroyDistributedObject(String name) {
//         containers.remove(name);
//     }
// 
//     @Override
//     public void beforeMigration(PartitionMigrationEvent partitionMigrationEvent) {
//     }
// 
//     @Override
//     public Operation prepareReplicationOperation(PartitionReplicationEvent event) {
// 
//         Map<String, byte[]> data = new HashMap<String, byte[]>();
//         int partitionId = event.getPartitionId();
//         for (String name : containers.keySet()) {
//             if (partitionId == getPartitionId(name)) {
//                 HLLWrapper number = containers.get(name);
//                 data.put(name, number.bytes());
//             }
//         }
//         return data.isEmpty() ? null : new HyperLogLogReplicationOperation(data);
//     }
// 
//     private int getPartitionId(String name) {
//         InternalPartitionService partitionService = nodeEngine.getPartitionService();
//         String partitionKey = getPartitionKey(name);
//         return partitionService.getPartitionId(partitionKey);
//     }
// 
//     @Override
//     public void commitMigration(PartitionMigrationEvent partitionMigrationEvent) {
//         if (partitionMigrationEvent.getMigrationEndpoint() == MigrationEndpoint.SOURCE) {
//             removePartition(partitionMigrationEvent.getPartitionId());
//         }
//     }
// 
//     @Override
//     public void rollbackMigration(PartitionMigrationEvent partitionMigrationEvent) {
//         if (partitionMigrationEvent.getMigrationEndpoint() == MigrationEndpoint.DESTINATION) {
//             removePartition(partitionMigrationEvent.getPartitionId());
//         }
//     }
// 
//     @Override
//     public void clearPartitionReplica(int partitionId) {
//         removePartition(partitionId);
//     }
// 
//     public void removePartition(int partitionId) {
//         final Iterator<String> iterator = containers.keySet().iterator();
//         while (iterator.hasNext()) {
//             String name = iterator.next();
//             if (getPartitionId(name) == partitionId) {
//                 iterator.remove();
//             }
//         }
//     }
// }
// 
// Path: src/main/java/org/rakam/util/HLLWrapper.java
// public class HLLWrapper {
//     final private static int SEED = 123456;
//     private HLL hll;
// 
//     public HLLWrapper() {
//         hll = new HLL(13/*log2m*/, 5/*registerWidth*/);
//     }
// 
//     public HLLWrapper(byte[] bytes) {
//         hll = HLL.fromBytes(bytes);
//     }
// 
//     public long cardinality() {
//         return hll.cardinality();
//     }
// 
//     public void union(HLLWrapper hll) {
//         this.hll.union(hll.hll);
//     }
// 
//     public void addAll(Collection<String> coll) {
//         for (String a : coll) {
//             byte[] s = a.getBytes();
//             hll.addRaw(MurmurHash3.murmurhash3x8632(s, 0, s.length, SEED));
//         }
//     }
// 
//     public void add(String obj) {
//         if (obj == null)
//             throw new IllegalArgumentException();
//         byte[] s = obj.getBytes();
// 
//         hll.addRaw(MurmurHash3.murmurhash3x8632(s, 0, s.length, SEED));
//     }
// 
//     public void set(byte[] bytes) {
//         hll = HLL.fromBytes(bytes);
//     }
// 
//     public byte[] bytes() {
//         return hll.toBytes();
//     }
// 
//     public void reset() {
//         hll = new HLL(13/*log2m*/, 5/*registerWidth*/);
//     }
// }


import com.hazelcast.nio.ObjectDataInput;
import com.hazelcast.nio.ObjectDataOutput;
import com.hazelcast.nio.serialization.IdentifiedDataSerializable;
import com.hazelcast.spi.AbstractOperation;
import org.rakam.cache.hazelcast.hyperloglog.HyperLogLogService;
import org.rakam.util.HLLWrapper;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
/**
 * Created by buremba on 10/07/14.
 */

package org.rakam.cache.hazelcast.hyperloglog.operations;



public class HyperLogLogReplicationOperation extends AbstractOperation implements IdentifiedDataSerializable {

    private Map<String, byte[]> migrationData;

    public HyperLogLogReplicationOperation() {
    }

    public HyperLogLogReplicationOperation(Map<String, byte[]> migrationData) {
        this.migrationData = migrationData;
    }

    @Override
    public void run() throws Exception {
// Path: src/main/java/org/rakam/cache/hazelcast/hyperloglog/HyperLogLogService.java
// public class HyperLogLogService implements ManagedService, RemoteService, MigrationAwareService {
// 
//     public static final String SERVICE_NAME = "rakam:hyperLogLogService";
// 
//     private NodeEngine nodeEngine;
//     private final ConcurrentMap<String, HLLWrapper> containers = new ConcurrentHashMap<String, HLLWrapper>();
//     private final ConstructorFunction<String, HLLWrapper> CountersConstructorFunction =
//             new ConstructorFunction<String, HLLWrapper>() {
//                 public HLLWrapper createNew(String key) {
//                     return new HLLWrapper();
//                 }
//             };
// 
//     public HyperLogLogService() {
//     }
// 
//     public HLLWrapper getHLL(String name) {
//         return getOrPutIfAbsent(containers, name, CountersConstructorFunction);
//     }
// 
//     @Override
//     public void init(NodeEngine nodeEngine, Properties properties) {
//         this.nodeEngine = nodeEngine;
//     }
// 
//     @Override
//     public void reset() {
//         containers.clear();
//     }
// 
//     @Override
//     public void shutdown(boolean terminate) {
//         reset();
//     }
// 
//     @Override
//     public HyperLogLogProxy createDistributedObject(String name) {
//         return new HyperLogLogProxy(name, nodeEngine, this);
//     }
// 
//     @Override
//     public void destroyDistributedObject(String name) {
//         containers.remove(name);
//     }
// 
//     @Override
//     public void beforeMigration(PartitionMigrationEvent partitionMigrationEvent) {
//     }
// 
//     @Override
//     public Operation prepareReplicationOperation(PartitionReplicationEvent event) {
// 
//         Map<String, byte[]> data = new HashMap<String, byte[]>();
//         int partitionId = event.getPartitionId();
//         for (String name : containers.keySet()) {
//             if (partitionId == getPartitionId(name)) {
//                 HLLWrapper number = containers.get(name);
//                 data.put(name, number.bytes());
//             }
//         }
//         return data.isEmpty() ? null : new HyperLogLogReplicationOperation(data);
//     }
// 
//     private int getPartitionId(String name) {
//         InternalPartitionService partitionService = nodeEngine.getPartitionService();
//         String partitionKey = getPartitionKey(name);
//         return partitionService.getPartitionId(partitionKey);
//     }
// 
//     @Override
//     public void commitMigration(PartitionMigrationEvent partitionMigrationEvent) {
//         if (partitionMigrationEvent.getMigrationEndpoint() == MigrationEndpoint.SOURCE) {
//             removePartition(partitionMigrationEvent.getPartitionId());
//         }
//     }
// 
//     @Override
//     public void rollbackMigration(PartitionMigrationEvent partitionMigrationEvent) {
//         if (partitionMigrationEvent.getMigrationEndpoint() == MigrationEndpoint.DESTINATION) {
//             removePartition(partitionMigrationEvent.getPartitionId());
//         }
//     }
// 
//     @Override
//     public void clearPartitionReplica(int partitionId) {
//         removePartition(partitionId);
//     }
// 
//     public void removePartition(int partitionId) {
//         final Iterator<String> iterator = containers.keySet().iterator();
//         while (iterator.hasNext()) {
//             String name = iterator.next();
//             if (getPartitionId(name) == partitionId) {
//                 iterator.remove();
//             }
//         }
//     }
// }
// 
// Path: src/main/java/org/rakam/util/HLLWrapper.java
// public class HLLWrapper {
//     final private static int SEED = 123456;
//     private HLL hll;
// 
//     public HLLWrapper() {
//         hll = new HLL(13/*log2m*/, 5/*registerWidth*/);
//     }
// 
//     public HLLWrapper(byte[] bytes) {
//         hll = HLL.fromBytes(bytes);
//     }
// 
//     public long cardinality() {
//         return hll.cardinality();
//     }
// 
//     public void union(HLLWrapper hll) {
//         this.hll.union(hll.hll);
//     }
// 
//     public void addAll(Collection<String> coll) {
//         for (String a : coll) {
//             byte[] s = a.getBytes();
//             hll.addRaw(MurmurHash3.murmurhash3x8632(s, 0, s.length, SEED));
//         }
//     }
// 
//     public void add(String obj) {
//         if (obj == null)
//             throw new IllegalArgumentException();
//         byte[] s = obj.getBytes();
// 
//         hll.addRaw(MurmurHash3.murmurhash3x8632(s, 0, s.length, SEED));
//     }
// 
//     public void set(byte[] bytes) {
//         hll = HLL.fromBytes(bytes);
//     }
// 
//     public byte[] bytes() {
//         return hll.toBytes();
//     }
// 
//     public void reset() {
//         hll = new HLL(13/*log2m*/, 5/*registerWidth*/);
//     }
// }


// Path: src/main/java/org/rakam/cache/hazelcast/hyperloglog/operations/HyperLogLogReplicationOperation.java
import com.hazelcast.nio.ObjectDataInput;
import com.hazelcast.nio.ObjectDataOutput;
import com.hazelcast.nio.serialization.IdentifiedDataSerializable;
import com.hazelcast.spi.AbstractOperation;
import org.rakam.cache.hazelcast.hyperloglog.HyperLogLogService;
import org.rakam.util.HLLWrapper;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * Created by buremba on 10/07/14.
 */

package org.rakam.cache.hazelcast.hyperloglog.operations;



public class HyperLogLogReplicationOperation extends AbstractOperation implements IdentifiedDataSerializable {

    private Map<String, byte[]> migrationData;

    public HyperLogLogReplicationOperation() {
    }

    public HyperLogLogReplicationOperation(Map<String, byte[]> migrationData) {
        this.migrationData = migrationData;
    }

    @Override
    public void run() throws Exception {
        HyperLogLogService service = getService();