unskip reliable topic tests

Author: mustafaiman

src/Address.ts

@@ -16,8 +16,8 @@
 
 import {IdentifiedDataSerializable} from './serialization/Serializable';
 import {DataInput, DataOutput} from './serialization/Data';
-import {ADDRESS_CLASS_ID, CLUSTER_DATA_FACTORY_ID} from './ClusterDataFactory';
 import * as net from 'net';
+import {ClusterDataFactoryHelper} from './ClusterDataFactoryHelper';
 
 class Address implements IdentifiedDataSerializable {
 
@@ -48,11 +48,11 @@ class Address implements IdentifiedDataSerializable {
     }
 
     getFactoryId(): number {
-        return CLUSTER_DATA_FACTORY_ID;
+        return ClusterDataFactoryHelper.FACTORY_ID;
     }
 
     getClassId(): number {
-        return ADDRESS_CLASS_ID;
+        return ClusterDataFactoryHelper.ADDRESS_ID;
     }
 
     equals(other: Address): boolean {

src/ClusterDataFactory.ts

@@ -16,15 +16,12 @@
 
 import {IdentifiedDataSerializableFactory, IdentifiedDataSerializable} from './serialization/Serializable';
 import Address = require('./Address');
-
-export const ADDRESS_CLASS_ID = 1;
-export const CLUSTER_DATA_FACTORY_ID = 0;
+import {ClusterDataFactoryHelper} from './ClusterDataFactoryHelper';
 
 export class ClusterDataFactory implements IdentifiedDataSerializableFactory {
 
     create(type: number): IdentifiedDataSerializable {
-
-        if (type === ADDRESS_CLASS_ID) {
+        if (type === ClusterDataFactoryHelper.ADDRESS_ID) {
             return new Address();
         }
 

src/ClusterDataFactoryHelper.ts

@@ -0,0 +1,20 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * 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.
+ */
+
+export class ClusterDataFactoryHelper {
+    static readonly FACTORY_ID = 0;
+    static readonly ADDRESS_ID = 1;
+}

src/proxy/RingbufferProxy.ts

@@ -91,7 +91,7 @@ export class RingbufferProxy<E> extends PartitionSpecificProxy implements IRingb
             .then<Array<E>>((raw: any) => {
                 return raw['items'].map((r: Data) => {
                     return this.toObject(r);
-                });
+                }, this);
             });
     }
 }

src/proxy/topic/RawTopicMessage.ts

@@ -23,8 +23,6 @@ export const RELIABLE_TOPIC_MESSAGE_FACTORY_ID = -18;
 export const RELIABLE_TOPIC_CLASS_ID = 2;
 
 export class RawTopicMessage implements IdentifiedDataSerializable {
-
-
     publishTime: Long;
     publisherAddress: Address;
     payload: Data;
@@ -52,11 +50,9 @@ export class RawTopicMessage implements IdentifiedDataSerializable {
 
 export class ReliableTopicMessageFactory implements IdentifiedDataSerializableFactory {
     create(type: number): IdentifiedDataSerializable {
-
         if (type === RELIABLE_TOPIC_CLASS_ID) {
             return new RawTopicMessage();
         }
-
         return null;
     }
 }

src/proxy/topic/ReliableTopicListenerRunner.ts

@@ -54,7 +54,7 @@ export class ReliableTopicListenerRunner<E> {
         this.ringbuffer.readMany(this.sequenceNumber, 1, this.batchSize).then((result: Array<RawTopicMessage>) => {
             if (!this.cancelled) {
                 result.forEach((raw: RawTopicMessage) => {
-                    var msg = new TopicMessage<E>();
+                    let msg = new TopicMessage<E>();
                     msg.messageObject = this.serializationService.toObject(raw.payload);
                     msg.publisher = raw.publisherAddress;
                     msg.publishingTime = raw.publishTime;
@@ -80,7 +80,7 @@ export class ReliableTopicListenerRunner<E> {
                 return;
             }
 
-            var message = 'Listener of topic "' + this.proxy.getName() + '" caught an exception, terminating listener';
+            var message = 'Listener of topic "' + this.proxy.getName() + '" caught an exception, terminating listener. ' + e;
             this.loggingService.warn('ReliableTopicListenerRunner', message);
 
             this.proxy.removeMessageListener(this.listenerId);

src/proxy/topic/TopicMessageListener.ts

@@ -13,7 +13,8 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+import {TopicMessage} from './TopicMessage';
 
 export interface TopicMessageListener<E> {
-    (item: E): void;
+    (message: TopicMessage<E>): void;
 }

src/serialization/SerializationService.ts

@@ -30,9 +30,9 @@ 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';
 import {RELIABLE_TOPIC_MESSAGE_FACTORY_ID, ReliableTopicMessageFactory} from '../proxy/topic/RawTopicMessage';
-import {CLUSTER_DATA_FACTORY_ID, ClusterDataFactory} from '../ClusterDataFactory';
+import {ClusterDataFactoryHelper} from '../ClusterDataFactoryHelper';
+import {ClusterDataFactory} from '../ClusterDataFactory';
 
 export interface SerializationService {
     toData(object: any, paritioningStrategy?: any) : Data;
@@ -257,7 +257,7 @@ export class SerializationServiceV1 implements SerializationService {
         }
         factories[PREDICATE_FACTORY_ID] = new PredicateFactory(DefaultPredicates);
         factories[RELIABLE_TOPIC_MESSAGE_FACTORY_ID] = new ReliableTopicMessageFactory();
-        factories[CLUSTER_DATA_FACTORY_ID] = new ClusterDataFactory();
+        factories[ClusterDataFactoryHelper.FACTORY_ID] = new ClusterDataFactory();
         this.registerSerializer('identified', new IdentifiedDataSerializableSerializer(factories));
     }
 

test/topic/TopicTest.js

@@ -24,7 +24,6 @@ var Controller = require('./../RC');
 var RawTopicMessage = require('../../lib/proxy/topic/RawTopicMessage').RawTopicMessage;
 var fs = require('fs');
 var Long = require('long');
-
 var Promise = require('bluebird');
 
 var createConfig = function () {
@@ -53,7 +52,7 @@ var generateItems = function (client, howMany) {
     }
     return all;
 };
-describe.skip("Reliable Topic Proxy", function () {
+describe("Reliable Topic Proxy", function () {
 
     var cluster;
     var clientOne;
@@ -68,7 +67,6 @@ describe.skip("Reliable Topic Proxy", function () {
             cluster = response;
             return Controller.startMember(cluster.id);
         }).then(function () {
-
             var config = createConfig();
 
             return Promise.all([
@@ -90,8 +88,9 @@ describe.skip("Reliable Topic Proxy", function () {
     });
 
     it("writes and reads messages", function (done) {
-        var topicOne = clientOne.getReliableTopic("t");
-        var topicTwo = clientTwo.getReliableTopic("t");
+        var topicName = 't' + Math.random();
+        var topicOne = clientOne.getReliableTopic(topicName);
+        var topicTwo = clientTwo.getReliableTopic(topicName);
 
         topicTwo.addMessageListener(function (msg) {
             if (msg.messageObject["value"] === "foo") {
@@ -105,15 +104,16 @@ describe.skip("Reliable Topic Proxy", function () {
     });
 
     it('removed message listener does not receive items after removal', function (done) {
-        var topicOne = clientOne.getReliableTopic("t");
-        var topicTwo = clientTwo.getReliableTopic("t");
+        var topicName = 't' + Math.random();
+        var topicOne = clientOne.getReliableTopic(topicName);
+        var topicTwo = clientTwo.getReliableTopic(topicName);
 
         var receivedMessages = 0;
 
         var id = topicTwo.addMessageListener(function (msg) {
             receivedMessages++;
             if (receivedMessages > 2) {
-                done(new Error('Keep receiving messages after removal.'));
+                done(new Error('Kept receiving messages after message listener is removed.'));
             }
         });
 
@@ -129,11 +129,11 @@ describe.skip("Reliable Topic Proxy", function () {
         }, 500);
     });
 
-    it("blocks when there is no more space", function (done) {
+    it("blocks when there is no more space", function () {
         var topic = clientOne.getReliableTopic("blocking");
         var ringbuffer = topic.getRingbuffer();
 
-        ringbuffer.capacity().then(function (capacity) {
+        return ringbuffer.capacity().then(function (capacity) {
             var all = [];
 
             for (var i = 0; i < capacity.toNumber() + 1; i++) {
@@ -142,19 +142,19 @@ describe.skip("Reliable Topic Proxy", function () {
 
             return ringbuffer.addAll(all);
         }).then(function () {
-            var startTime = new Date().getTime();
-            topic.publish(-50).then(function () {
+            var startTime = Date.now();
+            return topic.publish(-50).then(function () {
                 /*
                  Here we check that the call was indeed blocking
                  until the TTL of the first inserted entry has passed
                  */
 
-                var elapsed = new Date().getTime() - startTime;
+                var elapsed = Date.now() - startTime;
 
                 if (elapsed > 2000) {
-                    done();
+                    return;
                 } else {
-                    done(new Error("Message was published too fast, expected at least a 2 second delay, got: " + elapsed));
+                    throw new Error("Message was published too fast, expected at least a 2 second delay, got: " + elapsed);
                 }
             });
         });
@@ -173,7 +173,6 @@ describe.skip("Reliable Topic Proxy", function () {
 
         var all = generateItems(clientOne, 20);
 
-
         ringbuffer.addAll(all);
     });
 
@@ -195,7 +194,6 @@ describe.skip("Reliable Topic Proxy", function () {
         });
     });
 
-
     it("overwrites the oldest item when there is no more space", function () {
         var topic = clientOne.getReliableTopic("overwrite");
         var ringbuffer = topic.getRingbuffer();