1.0.73

package-lock.json

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

package.json

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

test/test.ts

@@ -74,7 +74,7 @@ tap.test('a second client should be able to subscibe', async () => {
 tap.test('should receive a message correctly', async () => {});
 
 tap.test('should disconnect the client correctly', async () => {
-  testClientUniverse.stop();
+  await testClientUniverse.stop();
 });
 
 tap.test('should end the server correctly', async tools => {

ts/interfaces/socketfunctionrequests.ts

@@ -15,6 +15,6 @@ export interface ISocketRequest_ProcessMessage {
   method: 'processMessage';
   request: interfaces.IUniverseMessage;
   response: {
-    messageStatus: 'ok'
+    messageStatus: 'ok' | 'channel not found'
   };
 }

ts/smartuniverse.classes.clientuniverse.ts

@@ -76,8 +76,8 @@ export class ClientUniverse {
     await this.checkConnection();
   }
 
-  public stop() {
-    this.smartsocketClient.disconnect();
+  public async stop() {
+    await this.smartsocketClient.disconnect();
   }
 
   /**
@@ -113,17 +113,30 @@ export class ClientUniverse {
       /**
        * handles message reception
        */
-      const socketFunctionProcessMessage = new plugins.smartsocket.SocketFunction<interfaces.ISocketRequest_ProcessMessage>({
+      const socketFunctionProcessMessage = new plugins.smartsocket.SocketFunction<
+        interfaces.ISocketRequest_ProcessMessage
+      >({
         funcName: 'processMessage',
         allowedRoles: [],
-        funcDef: async (messageDescriptorArg) => {
+        funcDef: async messageDescriptorArg => {
           plugins.smartlog.defaultLogger.log('info', 'Got message from server');
-          this.observableIntake.push(
-            ClientUniverseMessage.createMessageFromMessageDescriptor(messageDescriptorArg)
+          const clientUniverseMessage = ClientUniverseMessage.createMessageFromMessageDescriptor(
+            messageDescriptorArg
           );
-          return {
-            messageStatus: 'ok'
-          };
+          this.observableIntake.push(clientUniverseMessage);
+
+          // lets find the corresponding channel
+          const targetChannel = this.getChannel(clientUniverseMessage.targetChannelName);
+          if (targetChannel) {
+            await targetChannel.emitMessageLocally(clientUniverseMessage);
+            return {
+              messageStatus: 'ok'
+            };
+          } else {
+            return {
+              messageStatus: 'channel not found'
+            };
+          }
         }
       });
 

ts/smartuniverse.classes.clientuniversechannel.ts

@@ -2,6 +2,7 @@ import * as plugins from './smartuniverse.plugins';
 import * as interfaces from './interfaces';
 
 import { ClientUniverse } from './';
+import { ClientUniverseMessage } from './smartuniverse.classes.clientuniversemessage';
 
 export class ClientUniverseChannel implements interfaces.IUniverseChannel {
   // ======
@@ -35,6 +36,7 @@ export class ClientUniverseChannel implements interfaces.IUniverseChannel {
   public name: string;
   public passphrase: string;
   public status: 'subscribed' | 'unsubscribed' = 'unsubscribed';
+  private subject = new plugins.smartrx.rxjs.Subject();
 
   // refs
   public clientUniverseRef: ClientUniverse;
@@ -49,7 +51,8 @@ export class ClientUniverseChannel implements interfaces.IUniverseChannel {
    * subscribes to a channel
    * tells the universe about this instances interest into a channel
    */
-  public async subscribe() {
+  public async subscribe(observerArg?: plugins.smartrx.rxjs.Observer<any>) {
+    // lets make sure the channel is connected
     if (this.status === 'unsubscribed') {
       const response = await this.clientUniverseRef.smartsocketClient.serverCall<interfaces.ISocketRequest_SubscribeChannel>(
         'subscribeChannel',
@@ -60,6 +63,15 @@ export class ClientUniverseChannel implements interfaces.IUniverseChannel {
       );
       this.status = response.subscriptionStatus;
     }
+
+    if (observerArg) {
+      return this.subject.subscribe(observerArg);
+    }
+
+  }
+
+  public async emitMessageLocally(messageArg: ClientUniverseMessage) {
+    
   }
 
   /**

ts/smartuniverse.classes.reactionrequest.ts

@@ -0,0 +1,5 @@
+import * as plugins from './smartuniverse.plugins';
+
+export class ReactionRequest {
+
+}

ts/smartuniverse.classes.reactionresponse.ts

@@ -0,0 +1,3 @@
+import * as plugins from './smartuniverse.plugins';
+
+export class ReactionResponse {}