Merge pull request #165 from mustafaiman/semaphore

asimarslan · web-flow · commit b0927dad0f54 · 2017-02-28T09:24:34.000+02:00
Semaphore
diff --git a/.gitignore b/.gitignore
@@ -24,7 +24,7 @@ npm-debug.log
 
 # TypeScript
 *.js.map
-lib/*
+lib/
 
 # TSDoc
 docs/
diff --git a/src/HazelcastClient.ts b/src/HazelcastClient.ts
@@ -23,6 +23,7 @@ import {IRingbuffer} from './proxy/IRingbuffer';
 import {ITopic} from './proxy/topic/ITopic';
 import {ReliableTopicProxy} from './proxy/topic/ReliableTopicProxy';
 import {IReplicatedMap} from './proxy/IReplicatedMap';
+import {ISemaphore} from './proxy/ISemaphore';
 import {IAtomicLong} from './proxy/IAtomicLong';
 
 export default class HazelcastClient {
@@ -187,6 +188,15 @@ export default class HazelcastClient {
         return <IAtomicLong>this.proxyManager.getOrCreateProxy(name, this.proxyManager.ATOMICLONG_SERVICE);
     }
 
+    /**
+     * Returns the distributed semaphore instance with given name.
+     * @param name
+     * @returns {ISemaphore}
+     */
+    getSemaphore(name: string): ISemaphore {
+        return <ISemaphore>this.proxyManager.getOrCreateProxy(name, this.proxyManager.SEMAPHORE_SERVICE);
+    }
+
     /**
      * Return configuration that this instance started with.
      * Returned configuration object should not be modified.
diff --git a/src/Util.ts b/src/Util.ts
@@ -21,6 +21,10 @@ export function shuffleArray<T>(array: Array<T>): void {
     }
 }
 
+export function assertNotNegative(v: number, message: string = 'The value cannot be negative.') {
+    assert(v >= 0, message);
+}
+
 export function getType(obj: any): string {
     assertNotNull(obj);
     if (Long.isLong(obj)) {
diff --git a/src/codec/SemaphoreAcquireCodec.ts b/src/codec/SemaphoreAcquireCodec.ts
@@ -0,0 +1,30 @@
+/* tslint:disable */
+import ClientMessage = require('../ClientMessage');
+import {BitsUtil} from '../BitsUtil';
+import {SemaphoreMessageType} from './SemaphoreMessageType';
+
+const REQUEST_TYPE = SemaphoreMessageType.SEMAPHORE_ACQUIRE;
+const RESPONSE_TYPE = 100;
+const RETRYABLE = false;
+
+export class SemaphoreAcquireCodec {
+
+    static calculateSize(name: string, permits: number) {
+// Calculates the request payload size
+        let dataSize: number = 0;
+        dataSize += BitsUtil.calculateSizeString(name);
+        dataSize += BitsUtil.INT_SIZE_IN_BYTES;
+        return dataSize;
+    }
+
+    static encodeRequest(name: string, permits: number) {
+// Encode request into clientMessage
+        const clientMessage = ClientMessage.newClientMessage(this.calculateSize(name, permits));
+        clientMessage.setMessageType(REQUEST_TYPE);
+        clientMessage.setRetryable(RETRYABLE);
+        clientMessage.appendString(name);
+        clientMessage.appendInt32(permits);
+        clientMessage.updateFrameLength();
+        return clientMessage;
+    }
+}
diff --git a/src/codec/SemaphoreAvailablePermitsCodec.ts b/src/codec/SemaphoreAvailablePermitsCodec.ts
@@ -0,0 +1,36 @@
+/* tslint:disable */
+import ClientMessage = require('../ClientMessage');
+import {BitsUtil} from '../BitsUtil';
+import {Data} from '../serialization/Data';
+import {SemaphoreMessageType} from './SemaphoreMessageType';
+
+const REQUEST_TYPE = SemaphoreMessageType.SEMAPHORE_AVAILABLEPERMITS;
+const RESPONSE_TYPE = 102;
+const RETRYABLE = true;
+
+export class SemaphoreAvailablePermitsCodec {
+
+    static calculateSize(name: string) {
+// Calculates the request payload size
+        let dataSize: number = 0;
+        dataSize += BitsUtil.calculateSizeString(name);
+        return dataSize;
+    }
+
+    static encodeRequest(name: string) {
+// Encode request into clientMessage
+        const clientMessage = ClientMessage.newClientMessage(this.calculateSize(name));
+        clientMessage.setMessageType(REQUEST_TYPE);
+        clientMessage.setRetryable(RETRYABLE);
+        clientMessage.appendString(name);
+        clientMessage.updateFrameLength();
+        return clientMessage;
+    }
+
+    static decodeResponse(clientMessage: ClientMessage, toObjectFunction: (data: Data) => any = null) {
+// Decode response from client message
+        const parameters: any = {'response': null};
+        parameters['response'] = clientMessage.readInt32();
+        return parameters;
+    }
+}
diff --git a/src/codec/SemaphoreDrainPermitsCodec.ts b/src/codec/SemaphoreDrainPermitsCodec.ts
@@ -0,0 +1,36 @@
+/* tslint:disable */
+import ClientMessage = require('../ClientMessage');
+import {BitsUtil} from '../BitsUtil';
+import {Data} from '../serialization/Data';
+import {SemaphoreMessageType} from './SemaphoreMessageType';
+
+const REQUEST_TYPE = SemaphoreMessageType.SEMAPHORE_DRAINPERMITS;
+const RESPONSE_TYPE = 102;
+const RETRYABLE = true;
+
+export class SemaphoreDrainPermitsCodec {
+
+    static calculateSize(name: string) {
+// Calculates the request payload size
+        let dataSize: number = 0;
+        dataSize += BitsUtil.calculateSizeString(name);
+        return dataSize;
+    }
+
+    static encodeRequest(name: string) {
+// Encode request into clientMessage
+        const clientMessage = ClientMessage.newClientMessage(this.calculateSize(name));
+        clientMessage.setMessageType(REQUEST_TYPE);
+        clientMessage.setRetryable(RETRYABLE);
+        clientMessage.appendString(name);
+        clientMessage.updateFrameLength();
+        return clientMessage;
+    }
+
+    static decodeResponse(clientMessage: ClientMessage, toObjectFunction: (data: Data) => any = null) {
+// Decode response from client message
+        const parameters: any = {'response': null};
+        parameters['response'] = clientMessage.readInt32();
+        return parameters;
+    }
+}
diff --git a/src/codec/SemaphoreInitCodec.ts b/src/codec/SemaphoreInitCodec.ts
@@ -0,0 +1,38 @@
+/* tslint:disable */
+import ClientMessage = require('../ClientMessage');
+import {BitsUtil} from '../BitsUtil';
+import {Data} from '../serialization/Data';
+import {SemaphoreMessageType} from './SemaphoreMessageType';
+
+const REQUEST_TYPE = SemaphoreMessageType.SEMAPHORE_INIT;
+const RESPONSE_TYPE = 101;
+const RETRYABLE = false;
+
+export class SemaphoreInitCodec {
+
+    static calculateSize(name: string, permits: number) {
+// Calculates the request payload size
+        let dataSize: number = 0;
+        dataSize += BitsUtil.calculateSizeString(name);
+        dataSize += BitsUtil.INT_SIZE_IN_BYTES;
+        return dataSize;
+    }
+
+    static encodeRequest(name: string, permits: number) {
+// Encode request into clientMessage
+        const clientMessage = ClientMessage.newClientMessage(this.calculateSize(name, permits));
+        clientMessage.setMessageType(REQUEST_TYPE);
+        clientMessage.setRetryable(RETRYABLE);
+        clientMessage.appendString(name);
+        clientMessage.appendInt32(permits);
+        clientMessage.updateFrameLength();
+        return clientMessage;
+    }
+
+    static decodeResponse(clientMessage: ClientMessage, toObjectFunction: (data: Data) => any = null) {
+// Decode response from client message
+        const parameters: any = {'response': null};
+        parameters['response'] = clientMessage.readBoolean();
+        return parameters;
+    }
+}
diff --git a/src/codec/SemaphoreMessageType.ts b/src/codec/SemaphoreMessageType.ts
@@ -0,0 +1,11 @@
+/* tslint:disable */
+
+export class SemaphoreMessageType {
+    static SEMAPHORE_INIT = 0x0d01;
+    static SEMAPHORE_ACQUIRE = 0x0d02;
+    static SEMAPHORE_AVAILABLEPERMITS = 0x0d03;
+    static SEMAPHORE_DRAINPERMITS = 0x0d04;
+    static SEMAPHORE_REDUCEPERMITS = 0x0d05;
+    static SEMAPHORE_RELEASE = 0x0d06;
+    static SEMAPHORE_TRYACQUIRE = 0x0d07;
+}
diff --git a/src/codec/SemaphoreReducePermitsCodec.ts b/src/codec/SemaphoreReducePermitsCodec.ts
@@ -0,0 +1,30 @@
+/* tslint:disable */
+import ClientMessage = require('../ClientMessage');
+import {BitsUtil} from '../BitsUtil';
+import {SemaphoreMessageType} from './SemaphoreMessageType';
+
+const REQUEST_TYPE = SemaphoreMessageType.SEMAPHORE_REDUCEPERMITS;
+const RESPONSE_TYPE = 100;
+const RETRYABLE = false;
+
+export class SemaphoreReducePermitsCodec {
+
+    static calculateSize(name: string, reduction: number) {
+// Calculates the request payload size
+        let dataSize: number = 0;
+        dataSize += BitsUtil.calculateSizeString(name);
+        dataSize += BitsUtil.INT_SIZE_IN_BYTES;
+        return dataSize;
+    }
+
+    static encodeRequest(name: string, reduction: number) {
+// Encode request into clientMessage
+        const clientMessage = ClientMessage.newClientMessage(this.calculateSize(name, reduction));
+        clientMessage.setMessageType(REQUEST_TYPE);
+        clientMessage.setRetryable(RETRYABLE);
+        clientMessage.appendString(name);
+        clientMessage.appendInt32(reduction);
+        clientMessage.updateFrameLength();
+        return clientMessage;
+    }
+}
diff --git a/src/codec/SemaphoreReleaseCodec.ts b/src/codec/SemaphoreReleaseCodec.ts
@@ -0,0 +1,30 @@
+/* tslint:disable */
+import ClientMessage = require('../ClientMessage');
+import {BitsUtil} from '../BitsUtil';
+import {SemaphoreMessageType} from './SemaphoreMessageType';
+
+const REQUEST_TYPE = SemaphoreMessageType.SEMAPHORE_RELEASE;
+const RESPONSE_TYPE = 100;
+const RETRYABLE = false;
+
+export class SemaphoreReleaseCodec {
+
+    static calculateSize(name: string, permits: number) {
+// Calculates the request payload size
+        let dataSize: number = 0;
+        dataSize += BitsUtil.calculateSizeString(name);
+        dataSize += BitsUtil.INT_SIZE_IN_BYTES;
+        return dataSize;
+    }
+
+    static encodeRequest(name: string, permits: number) {
+// Encode request into clientMessage
+        const clientMessage = ClientMessage.newClientMessage(this.calculateSize(name, permits));
+        clientMessage.setMessageType(REQUEST_TYPE);
+        clientMessage.setRetryable(RETRYABLE);
+        clientMessage.appendString(name);
+        clientMessage.appendInt32(permits);
+        clientMessage.updateFrameLength();
+        return clientMessage;
+    }
+}
diff --git a/src/codec/SemaphoreTryAcquireCodec.ts b/src/codec/SemaphoreTryAcquireCodec.ts
@@ -0,0 +1,41 @@
+/* tslint:disable */
+import ClientMessage = require('../ClientMessage');
+import {BitsUtil} from '../BitsUtil';
+import {Data} from '../serialization/Data';
+import {SemaphoreMessageType} from './SemaphoreMessageType';
+import Long = require('long');
+
+const REQUEST_TYPE = SemaphoreMessageType.SEMAPHORE_TRYACQUIRE;
+const RESPONSE_TYPE = 101;
+const RETRYABLE = false;
+
+export class SemaphoreTryAcquireCodec {
+
+    static calculateSize(name: string, permits: number, timeout: Long|number|string) {
+// Calculates the request payload size
+        let dataSize: number = 0;
+        dataSize += BitsUtil.calculateSizeString(name);
+        dataSize += BitsUtil.INT_SIZE_IN_BYTES;
+        dataSize += BitsUtil.LONG_SIZE_IN_BYTES;
+        return dataSize;
+    }
+
+    static encodeRequest(name: string, permits: number, timeout: Long|number|string) {
+// Encode request into clientMessage
+        const clientMessage = ClientMessage.newClientMessage(this.calculateSize(name, permits, timeout));
+        clientMessage.setMessageType(REQUEST_TYPE);
+        clientMessage.setRetryable(RETRYABLE);
+        clientMessage.appendString(name);
+        clientMessage.appendInt32(permits);
+        clientMessage.appendLong(timeout);
+        clientMessage.updateFrameLength();
+        return clientMessage;
+    }
+
+    static decodeResponse(clientMessage: ClientMessage, toObjectFunction: (data: Data) => any = null) {
+// Decode response from client message
+        const parameters: any = {'response': null};
+        parameters['response'] = clientMessage.readBoolean();
+        return parameters;
+    }
+}
diff --git a/src/proxy/ISemaphore.ts b/src/proxy/ISemaphore.ts
@@ -0,0 +1,58 @@
+import * as Promise from 'bluebird';
+import Long = require('long');
+import {DistributedObject} from '../DistributedObject';
+
+export interface ISemaphore extends DistributedObject {
+    /**
+     * Try to initialize this ISemaphore instance with the given permit count.
+     * @param permits
+     */
+    init(permits: number): Promise<boolean>;
+
+    /**
+     * Acquires the given number of permits if they are available, and returns
+     * immediately, reducing the number of available permits by the given amount.
+     * If insufficient permits are available returned promise is not resolved
+     * until until there are sufficient number of available permits or this
+     * {ISemaphore} is destroyed.
+     * @param permits
+     */
+    acquire(permits?: number): Promise<void>;
+
+    /**
+     * Returns the current number of permits currently available in this semaphore.
+     * This message is idempotent.
+     */
+    availablePermits(): Promise<number>;
+
+    /**
+     * Acquires and returns all permits that are immediately available.
+     */
+    drainPermits(): Promise<number>;
+
+    /**
+     * Shrinks the number of available permits by the indicated reduction.
+     * @param reduction
+     */
+    reducePermits(reduction: number): Promise<void>;
+
+    /**
+     * Releases the given number of permits, increasing the number of available
+     * permits by that amount. There is no requirement that a thread that releases a
+     * permit must have acquired that permit by calling one of the acquire()acquire
+     * methods. Correct usage of a semaphore is established by programming convention
+     * in the application
+     */
+    release(permits?: number): Promise<void>;
+
+    /**
+     * Acquires the given number of permits, if they are available, and returns
+     * immediately, with the value true, reducing the number of available permits
+     * by the given amount. If insufficient permits are available then this
+     * method will return immediately with the value false and the number of
+     * available permits is unchanged.
+     * @param permits
+     * @param timeout
+     */
+    tryAcquire(permits: number, timeout: Long | number): Promise<boolean>;
+}
diff --git a/src/proxy/ProxyManager.ts b/src/proxy/ProxyManager.ts
@@ -16,6 +16,7 @@ import {MultiMapProxy} from './MultiMapProxy';
 import {RingbufferProxy} from './RingbufferProxy';
 import {ReplicatedMapProxy} from './ReplicatedMapProxy';
 import {NearCachedMapProxy} from './NearCachedMapProxy';
+import {SemaphoreProxy} from './SemaphoreProxy';
 import {AtomicLongProxy} from './AtomicLongProxy';
 
 class ProxyManager {
@@ -27,6 +28,7 @@ class ProxyManager {
     public MULTIMAP_SERVICE: string = 'hz:impl:multiMapService';
     public RINGBUFFER_SERVICE: string = 'hz:impl:ringbufferService';
     public REPLICATEDMAP_SERVICE: string = 'hz:impl:replicatedMapService';
+    public SEMAPHORE_SERVICE: string = 'hz:impl:semaphoreService';
     public ATOMICLONG_SERVICE: string = 'hz:impl:atomicLongService';
 
     public service: any = {
@@ -38,6 +40,7 @@ class ProxyManager {
         'hz:impl:multiMapService': MultiMapProxy,
         'hz:impl:ringbufferService': RingbufferProxy,
         'hz:impl:replicatedMapService': ReplicatedMapProxy,
+        'hz:impl:semaphoreService': SemaphoreProxy
         'hz:impl:atomicLongService': AtomicLongProxy
     };
 
diff --git a/src/proxy/SemaphoreProxy.ts b/src/proxy/SemaphoreProxy.ts
diff --git a/test/semaphore/SemaphoreProxyTest.js b/test/semaphore/SemaphoreProxyTest.js

Original file line number	Diff line number	Diff line change
`@@ -21,6 +21,10 @@ export function shuffleArray<T>(array: Array<T>): void {`
`21`	`21`	`}`
`22`	`22`	`}`
`23`	`23`
	`24`	`+export function assertNotNegative(v: number, message: string = 'The value cannot be negative.') {`
	`25`	`+ assert(v >= 0, message);`
	`26`	`+}`
	`27`	`+`
`24`	`28`	`export function getType(obj: any): string {`
`25`	`29`	`assertNotNull(obj);`
`26`	`30`	`if (Long.isLong(obj)) {`