1.0.87

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@pushrocks/smartuniverse",
-  "version": "1.0.83",
+  "version": "1.0.87",
   "lockfileVersion": 1,
   "requires": true,
   "dependencies": {

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@pushrocks/smartuniverse",
-  "version": "1.0.83",
+  "version": "1.0.87",
   "private": false,
   "description": "messaging service for your micro services",
   "main": "dist/index.js",

test/test.ts

@@ -77,8 +77,10 @@ tap.test('should receive a message correctly', async (tools) => {
   const testChannel = testClientUniverse.getChannel(testChannelData.channelName);
   const testChannel2 = testClientUniverse2.getChannel(testChannelData.channelName);
   const subscription = testChannel2.subscribe(messageArg => {
+    if (messageArg.messageText === 'hellothere') {
       console.log('Yay##########');
       done.resolve();
+    }
   });
   await testChannel.sendMessage({
     messageText: 'hellothere'
@@ -86,11 +88,43 @@ tap.test('should receive a message correctly', async (tools) => {
   await done.promise;
 });
 
-tap.test('should disconnect the client correctly', async () => {
+interface IDemoReqRes {
-  await testClientUniverse.stop();
+  method: 'demo',
+  request: {
+    wowso: string;
+  };
+  response: {
+    hereso: string;
+  };
+}
+
+tap.test('ReactionRequest and ReactionResponse should work', async () => {
+  const reactionResponse = new smartuniverse.ReactionResponse<IDemoReqRes>({
+    channels: [testUniverse.getChannel(testChannelData.channelName)],
+    funcDef: async reqData => {
+      console.log(reqData);
+      return {
+        hereso: 'Hello there'
+      };
+    },
+    method: 'demo'
+  });
+  const reactionRequest = new smartuniverse.ReactionRequest<IDemoReqRes>({
+    method: 'demo'
+  });
+  const reactionResult = await reactionRequest.fire([testClientUniverse2.getChannel(testChannelData.channelName)], {
+    wowso: 'wowza'
+  });
+  const result = await reactionResult.getFirstResult();
+  console.log(result);
 });
 
-tap.test('should end the server correctly', async tools => {
+tap.test('should disconnect the client correctly', async (tools) => {
+  await testClientUniverse.stop();
+  await testClientUniverse2.stop();
+});
+
+tap.test('should end the server correctly', async (tools) => {
   await testUniverse.stopServer();
 });
 

ts/smartuniverse.classes.reactionrequest.ts

@@ -28,7 +28,7 @@ export class ReactionRequest<T extends plugins.typedrequestInterfaces.ITypedRequ
     this.method = optionsArg.method;
   }
 
-  public async fire(channelsArg: Array<UniverseChannel | ClientUniverseChannel>, requestDataArg: T['request'], timeoutMillisArg=60000) {
+  public async fire(channelsArg: Array<UniverseChannel | ClientUniverseChannel>, requestDataArg: T['request'], timeoutMillisArg=5000) {
     const subscriptionMap = new plugins.lik.Objectmap<plugins.smartrx.rxjs.Subscription>();
     const reactionResult = new ReactionResult<T>();
     const requestId = plugins.smartunique.shortId();
@@ -49,7 +49,7 @@ export class ReactionRequest<T extends plugins.typedrequestInterfaces.ITypedRequ
           request: requestDataArg,
           response: null
         }
-      }
+      };
       channel.sendMessage({
         messageText: 'reactionRequest',
         payload

ts/smartuniverse.classes.reactionresult.ts

@@ -2,7 +2,7 @@ import * as plugins from './smartuniverse.plugins';
 import { ReactionResponse } from './smartuniverse.classes.reactionresponse';
 
 export class ReactionResult<T extends plugins.typedrequestInterfaces.ITypedRequest> {
-  private resultSubject = new plugins.smartrx.rxjs.Subject<T['response']>();
+  private resultReplaySubject = new plugins.smartrx.rxjs.ReplaySubject<T['response']>();
   private endResult: Array<T['response']> = [];
   private completeDeferred = plugins.smartpromise.defer<Array<T['response']>>();
 
@@ -13,19 +13,34 @@ export class ReactionResult<T extends plugins.typedrequestInterfaces.ITypedReque
   }
 
   public resultSubscribe(observerArg: (responseArg: T['response']) => void) {
-    return this.resultSubject.subscribe(observerArg);
+    return this.resultReplaySubject.subscribe(observerArg);
   }
 
+  /**
+   * gets the end result as an array of all results
+   */
   public async getEndResult() {
     const result = await this.completeDeferred.promise;
     return result;
   }
 
+  /**
+   * if there is a single respondant, or you are only interested in the first result
+   */
+  public async getFirstResult() {
+    const done = plugins.smartpromise.defer<T['response']>();
+    const subscription = this.resultReplaySubject.subscribe(result => {
+      done.resolve(result);
+      subscription.unsubscribe();
+    });
+    return await done.promise;
+  }
+
   /**
    * push a reactionResponse
    */
   public async pushReactionResponse(responseArg: T['response']) {
-    this.resultSubject.next(responseArg);
+    this.resultReplaySubject.next(responseArg);
   }
   
   /**

ts/smartuniverse.classes.universe.ts

@@ -68,7 +68,7 @@ export class Universe {
   /**
    * returns a channel
    */
-  public getChannelByName(channelNameArg: string) {
+  public getChannel(channelNameArg: string) {
     return this.universeCache.channelMap.find(channelArg => {
       return channelArg.name === channelNameArg;
     });
@@ -178,6 +178,7 @@ export class Universe {
    * stop everything
    */
   public async stopServer() {
+    console.log('hi');
     await this.smartsocket.stop();
     if (!this.options.externalServer) {
       await this.smartexpressServer.stop();

ts/smartuniverse.classes.universecache.ts

@@ -19,7 +19,7 @@ export class UniverseCache {
   // INSTANCE
   // ========
   public standardMessageExpiry: number;
-  public destructionTime: number = 60000;
+  public destructionTime: number = 10000;
 
   /**
    * stores messages for this instance

ts/smartuniverse.classes.universechannel.ts

@@ -163,6 +163,6 @@ export class UniverseChannel {
       passphrase: this.passphrase,
       timestamp: Date.now()
     });
-    this.push(messageToSend);
+    this.universeRef.universeCache.addMessage(messageToSend);
   }
 }

ts/smartuniverse.classes.universemessage.ts

@@ -64,7 +64,7 @@ export class UniverseMessage<T> implements interfaces.IUniverseMessage {
     this.passphrase = messageDescriptor.passphrase;
     this.payload = messageDescriptor.payload;
     // prevent memory issues
-    this.fallBackDestruction();
+    this.setDestructionTimer();
   }
 
   public setUniverseCache(universeCacheArg: UniverseCache) {
@@ -73,17 +73,23 @@ export class UniverseMessage<T> implements interfaces.IUniverseMessage {
 
   public setTargetChannel() {}
 
-  public setDestructionTimer(selfdestructAfterArg: number) {
+  public setDestructionTimer(selfdestructAfterArg?: number) {
     if (selfdestructAfterArg) {
       this.destructionTimer = new Timer(selfdestructAfterArg);
       this.destructionTimer.start();
-
       // set up self destruction by removing this from the parent messageCache
       this.destructionTimer.completed.then(async () => {
         this.universeCache.messageMap.remove(this);
+      }).catch(err => {
+        console.log(err);
+        console.log(this);
       });
     } else {
-      this.fallBackDestruction();
+      plugins.smartdelay.delayFor(1000).then(() => {
+        if (!this.destructionTimer) {
+          this.setDestructionTimer(6000);
+        }
+      });
     }
   }
 
@@ -93,15 +99,4 @@ export class UniverseMessage<T> implements interfaces.IUniverseMessage {
   public handleAsBadMessage() {
     plugins.smartlog.defaultLogger.log('warn', 'received a bad message');
   }
-
-  /**
-   * prevents memory leaks if channels have no default
-   */
-  private fallBackDestruction() {
-    plugins.smartdelay.delayFor(1000).then(() => {
-      if (!this.destructionTimer) {
-        this.setDestructionTimer(6000);
-      }
-    });
-  }
 }