Merge pull request #137 from mustafaiman/partitionAware

PartitionAware

Author: GhostInAMachine

src/core/PartitionAware.ts

@@ -0,0 +1,24 @@
+/**
+ * Implementing PartitionAware allows one to override the default partitioning scheme.
+ * Instead of using the keys themselves to spread the data around the cluster the
+ * key returned by {@link getPartitionKey} is used.
+ * <p/>
+ * This provides the user with an ability to contain related keys within the same
+ * partition and, consequently, within the same node.
+ * <p/>
+ * In Hazelcast, disparate data structures will be stored on the same partition,
+ * based on the partition key. For example, if "Steve" was used, then the following would be on one partition.
+ * <ul>
+ *     <li>a customers IMap with an entry of key "Steve"</li>
+ *     <li>an orders IMap using a customer key type implementing PartitionAware with key "Steve</li>
+ *     <li>any queue named "Steve"</li>
+ *     <li>any PartitionAware object with partition key "Steve"</li>
+ * </ul>
+ */
+export interface PartitionAware<T> {
+    /**
+     * The key that will be used by Hazelcast to specify the partition.
+     * You should give the same key for objects that you want to be in the same partition.
+     */
+    getPartitionKey(): T;
+}

src/serialization/SerializationService.ts

@@ -14,6 +14,7 @@ import {PortableSerializer} from './portable/PortableSerializer';
 import {IdentifiedDataSerializableFactory} from './Serializable';
 import * as DefaultPredicates from './DefaultPredicates';
 import {PredicateFactory, PREDICATE_FACTORY_ID} from './PredicateFactory';
+import {PartitionAware} from '../core/PartitionAware';
 
 export interface SerializationService {
     toData(object: any, paritioningStrategy?: any) : Data;
@@ -50,7 +51,14 @@ export class SerializationServiceV1 implements SerializationService {
     toData(object: any, partitioningStrategy: any = this.defaultPartitionStrategy): Data {
         var dataOutput: DataOutput = new PositionalObjectDataOutput(1, this, this.serialiationConfig.isBigEndian);
         var serializer = this.findSerializerFor(object);
-        dataOutput.writeIntBE(this.calculatePartitionHash(object, partitioningStrategy));
+        //Check if object is partition aware
+        if (object != null && object.getPartitionKey) {
+            var partitionKey = object.getPartitionKey();
+            var serializedPartitionKey = this.toData(partitionKey);
+            dataOutput.writeIntBE(this.calculatePartitionHash(serializedPartitionKey, partitioningStrategy));
+        } else {
+            dataOutput.writeIntBE(this.calculatePartitionHash(object, partitioningStrategy));
+        }
         dataOutput.writeIntBE(serializer.getId());
         serializer.write(dataOutput, object);
         return new HeapData(dataOutput.toBuffer());

test/map/MapPartitionAwareTest.js

@@ -0,0 +1,99 @@
+var Client = require('../../.').Client;
+var Controller = require('../RC');
+var expect = require('chai').expect;
+var Promise = require('bluebird');
+
+describe('Map Partition Aware', function() {
+
+    var cluster;
+    var numOfEntries = 10000;
+    var memberCount = 3;
+    var members = [];
+    var client;
+    var map;
+    var mapName = 'testMap';
+
+    function PartitionAwareKey(key, partitionKey) {
+        this.key = key;
+        this.partitionKey = partitionKey;
+    }
+
+    PartitionAwareKey.prototype.getPartitionKey = function() {
+        return this.partitionKey;
+    };
+
+    function getLocalMapStats(serverInstance) {
+        return 'function getLocalMapStats() {' +
+            '   var map = instance_' + serverInstance + '.getMap("' + mapName + '");' +
+            '   return map.getLocalMapStats().getOwnedEntryCount();' +
+            '}' +
+            'result=""+getLocalMapStats();';
+    }
+
+    function _fillMap(map, ssize) {
+        var entryList = [];
+        for (var i = 0; i < ssize; i++) {
+            entryList.push([new PartitionAwareKey(''+Math.random(), 'specificKey'), ''+Math.random()]);
+        }
+        return map.putAll(entryList);
+    }
+
+    before(function() {
+        expect(memberCount, 'This test should have at least 2 members.').to.be.at.least(2);
+        this.timeout(20000);
+        return Controller.createCluster(null, null).then(function(c) {
+            cluster = c;
+            for (var i = 0; i < memberCount; i++) {
+                members.push(Controller.startMember(cluster.id));
+            }
+            return Promise.all(members);
+        }).then(function(m) {
+            members = m;
+            return Client.newHazelcastClient();
+        }).then(function(cl) {
+            client = cl;
+        });
+    });
+
+    after(function() {
+        this.timeout(20000);
+        client.shutdown();
+        return Controller.shutdownCluster(cluster.id);
+    });
+
+    beforeEach(function() {
+        map = client.getMap(mapName);
+    });
+
+    afterEach(function() {
+        return map.destroy();
+    });
+
+    it('put', function() {
+        this.timeout(10000);
+        return _fillMap(map, numOfEntries).then(function(newVal) {
+            var promises = members.map(function(member, index) {
+                return Controller.executeOnController(cluster.id, getLocalMapStats(index), 1);
+            });
+            return Promise.all(promises);
+        }).then(function(stats) {
+            var entriesPerMember = stats.map(function(item) {
+                return Number(item.result);
+            });
+            var expectedArray = [numOfEntries];
+            for (var i = 0; i < memberCount - 1; i++) {
+                expectedArray.push(0);
+            }
+            return expect(entriesPerMember, 'One member should have all of the entries. The rest will have 0 entries.').to.have.members(expectedArray);
+        });
+    });
+
+    it('get', function() {
+        var key = new PartitionAwareKey('key', 'partKey');
+        return map.put(key, 'value').then(function() {
+            return map.get(key);
+        }).then(function(val) {
+            return expect(val).to.equal('value');
+        });
+    })
+});