// Path: src/main/java/org/rakam/cache/hazelcast/treemap/OrderedCounterMap.java
// public class OrderedCounterMap implements Iterable<Counter>, DataSerializable {
//     Set<Counter> set = Collections.newSetFromMap(new ConcurrentSkipListMap());
//     Map<String, Counter> map = new ConcurrentHashMap<String, Counter>();
// 
//     public long increment(String key, long by) {
//         Counter c = map.get(key);
// 
//         if(c!=null) {
//             return map.get(key).increment();
//         }else {
//             c = new Counter(key, by);
//             map.put(c.id, c);
//             set.add(c);
//         }
//         return by;
//     }
// 
//     public Map<String, Long> getAll() {
//         Map<String, Long> ret = new LinkedHashMap(set.size());
//         Iterator<Counter> it = set.iterator();
//         while(it.hasNext()) {
//             Counter next = it.next();
//             ret.put(next.id, next.get());
//         }
//         return ret;
//     }
// 
//     public Map<String, Long> getTopItems(int numberOfItems) {
//         Map<String, Long> ret = new LinkedHashMap(set.size());
//         Iterator<Counter> it = set.iterator();
//         int i = 0;
//         while(it.hasNext() && numberOfItems>i) {
//             Counter next = it.next();
//             ret.put(next.id, next.get());
//             i++;
//         }
//         return ret;
//     }
// 
// 
//     @Override
//     public Iterator<Counter> iterator() {
//         return set.iterator();
//     }
// 
//     @Override
//     public void writeData(ObjectDataOutput out) throws IOException {
//         out.writeInt(set.size());
//         Iterator<Counter> it = set.iterator();
//         while(it.hasNext()) {
//             out.writeObject(it.next());
//         }
//     }
// 
//     @Override
//     public void readData(ObjectDataInput in) throws IOException {
//         int len = in.readInt();
//         for(int i=0; i<len; i++) {
//             Counter c = in.readObject();
//             map.put(c.id, c);
//             set.add(c);
//         }
//     }
// }
// 
// Path: src/main/java/org/rakam/cache/hazelcast/treemap/TreeMapService.java
// public class TreeMapService implements ManagedService, RemoteService, MigrationAwareService {
// 
//     public static final String SERVICE_NAME = "rakam:treeMapService";
// 
//     private NodeEngine nodeEngine;
//     private final ConcurrentMap<String, OrderedCounterMap> containers = new ConcurrentHashMap<String, OrderedCounterMap>();
//     private final ConstructorFunction<String, OrderedCounterMap> CountersConstructorFunction =
//             new ConstructorFunction<String, OrderedCounterMap>() {
//                 public OrderedCounterMap createNew(String key) {
//                     return new OrderedCounterMap();
//                 }
//             };
// 
//     public TreeMapService() {
//     }
// 
//     public OrderedCounterMap getHLL(String name) {
//         return getOrPutIfAbsent(containers, name, CountersConstructorFunction);
//     }
// 
//     public void setHLL(String name, OrderedCounterMap map) {
//         containers.put(name, map);
//     }
// 
//     @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 TreeMapProxy createDistributedObject(String name) {
//         return new TreeMapProxy(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, OrderedCounterMap> data = new HashMap();
//         int partitionId = event.getPartitionId();
//         for (String name : containers.keySet()) {
//             if (partitionId == getPartitionId(name)) {
//                 OrderedCounterMap number = containers.get(name);
//                 data.put(name, number);
//             }
//         }
//         return data.isEmpty() ? null : new TreeMapReplicationOperation(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();
//             }
//         }
//     }
// }


import com.hazelcast.spi.BackupOperation;
import org.rakam.cache.hazelcast.treemap.OrderedCounterMap;
import org.rakam.cache.hazelcast.treemap.TreeMapService;
package org.rakam.cache.hazelcast.treemap.operations;


/**
 * Created by buremba <Burak Emre Kabakcı> on 15/07/14 03:23.
 */
public class IncrementByBackupOperation extends TreeMapBaseOperation implements BackupOperation {
    private long by;

    public IncrementByBackupOperation(String name, long by) {
        super(name);
        this.by = by;
    }

    public IncrementByBackupOperation() {

    }

    @Override
    public void run() throws Exception {
// Path: src/main/java/org/rakam/cache/hazelcast/treemap/OrderedCounterMap.java
// public class OrderedCounterMap implements Iterable<Counter>, DataSerializable {
//     Set<Counter> set = Collections.newSetFromMap(new ConcurrentSkipListMap());
//     Map<String, Counter> map = new ConcurrentHashMap<String, Counter>();
// 
//     public long increment(String key, long by) {
//         Counter c = map.get(key);
// 
//         if(c!=null) {
//             return map.get(key).increment();
//         }else {
//             c = new Counter(key, by);
//             map.put(c.id, c);
//             set.add(c);
//         }
//         return by;
//     }
// 
//     public Map<String, Long> getAll() {
//         Map<String, Long> ret = new LinkedHashMap(set.size());
//         Iterator<Counter> it = set.iterator();
//         while(it.hasNext()) {
//             Counter next = it.next();
//             ret.put(next.id, next.get());
//         }
//         return ret;
//     }
// 
//     public Map<String, Long> getTopItems(int numberOfItems) {
//         Map<String, Long> ret = new LinkedHashMap(set.size());
//         Iterator<Counter> it = set.iterator();
//         int i = 0;
//         while(it.hasNext() && numberOfItems>i) {
//             Counter next = it.next();
//             ret.put(next.id, next.get());
//             i++;
//         }
//         return ret;
//     }
// 
// 
//     @Override
//     public Iterator<Counter> iterator() {
//         return set.iterator();
//     }
// 
//     @Override
//     public void writeData(ObjectDataOutput out) throws IOException {
//         out.writeInt(set.size());
//         Iterator<Counter> it = set.iterator();
//         while(it.hasNext()) {
//             out.writeObject(it.next());
//         }
//     }
// 
//     @Override
//     public void readData(ObjectDataInput in) throws IOException {
//         int len = in.readInt();
//         for(int i=0; i<len; i++) {
//             Counter c = in.readObject();
//             map.put(c.id, c);
//             set.add(c);
//         }
//     }
// }
// 
// Path: src/main/java/org/rakam/cache/hazelcast/treemap/TreeMapService.java
// public class TreeMapService implements ManagedService, RemoteService, MigrationAwareService {
// 
//     public static final String SERVICE_NAME = "rakam:treeMapService";
// 
//     private NodeEngine nodeEngine;
//     private final ConcurrentMap<String, OrderedCounterMap> containers = new ConcurrentHashMap<String, OrderedCounterMap>();
//     private final ConstructorFunction<String, OrderedCounterMap> CountersConstructorFunction =
//             new ConstructorFunction<String, OrderedCounterMap>() {
//                 public OrderedCounterMap createNew(String key) {
//                     return new OrderedCounterMap();
//                 }
//             };
// 
//     public TreeMapService() {
//     }
// 
//     public OrderedCounterMap getHLL(String name) {
//         return getOrPutIfAbsent(containers, name, CountersConstructorFunction);
//     }
// 
//     public void setHLL(String name, OrderedCounterMap map) {
//         containers.put(name, map);
//     }
// 
//     @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 TreeMapProxy createDistributedObject(String name) {
//         return new TreeMapProxy(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, OrderedCounterMap> data = new HashMap();
//         int partitionId = event.getPartitionId();
//         for (String name : containers.keySet()) {
//             if (partitionId == getPartitionId(name)) {
//                 OrderedCounterMap number = containers.get(name);
//                 data.put(name, number);
//             }
//         }
//         return data.isEmpty() ? null : new TreeMapReplicationOperation(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/cache/hazelcast/treemap/operations/IncrementByBackupOperation.java
import com.hazelcast.spi.BackupOperation;
import org.rakam.cache.hazelcast.treemap.OrderedCounterMap;
import org.rakam.cache.hazelcast.treemap.TreeMapService;

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


/**
 * Created by buremba <Burak Emre Kabakcı> on 15/07/14 03:23.
 */
public class IncrementByBackupOperation extends TreeMapBaseOperation implements BackupOperation {
    private long by;

    public IncrementByBackupOperation(String name, long by) {
        super(name);
        this.by = by;
    }

    public IncrementByBackupOperation() {

    }

    @Override
    public void run() throws Exception {
        OrderedCounterMap map = ((TreeMapService) getService()).getHLL(name);