1.0.93

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@pushrocks/smartuniverse",
-  "version": "1.0.88",
+  "version": "1.0.93",
   "private": false,
   "description": "messaging service for your micro services",
   "main": "dist/index.js",
@@ -15,10 +15,10 @@
   },
   "devDependencies": {
     "@gitzone/tsbuild": "^2.1.17",
-    "@gitzone/tstest": "^1.0.24",
+    "@gitzone/tstest": "^1.0.28",
     "@pushrocks/tapbundle": "^3.0.13",
-    "@types/node": "^12.7.4",
-    "tslint": "^5.20.0",
+    "@types/node": "^12.12.6",
+    "tslint": "^5.20.1",
     "tslint-config-prettier": "^1.18.0"
   },
   "peerDependencies": {
@@ -27,15 +27,15 @@
   "dependencies": {
     "@apiglobal/typedrequest-interfaces": "^1.0.7",
     "@pushrocks/lik": "^3.0.11",
-    "@pushrocks/smartdelay": "^2.0.3",
-    "@pushrocks/smartexpress": "^3.0.40",
-    "@pushrocks/smartfile": "^7.0.4",
+    "@pushrocks/smartdelay": "^2.0.6",
+    "@pushrocks/smartexpress": "^3.0.52",
+    "@pushrocks/smartfile": "^7.0.6",
     "@pushrocks/smarthash": "^2.0.6",
-    "@pushrocks/smartlog": "^2.0.19",
-    "@pushrocks/smartpromise": "^3.0.2",
-    "@pushrocks/smartrequest": "^1.1.27",
+    "@pushrocks/smartlog": "^2.0.21",
+    "@pushrocks/smartpromise": "^3.0.6",
+    "@pushrocks/smartrequest": "^1.1.42",
     "@pushrocks/smartrx": "^2.0.5",
-    "@pushrocks/smartsocket": "^1.1.49",
+    "@pushrocks/smartsocket": "^1.1.58",
     "@pushrocks/smarttime": "^3.0.12",
     "@pushrocks/smartunique": "^3.0.1"
   },

ts/smartuniverse.classes.clientuniverse.ts

@@ -19,9 +19,9 @@ export interface IClientOptions {
  * allows connecting to a universe server
  */
 export class ClientUniverse {
-  public options;
+  public options: IClientOptions;
   public smartsocketClient: plugins.smartsocket.SmartsocketClient;
-  public observableIntake: plugins.smartrx.ObservableIntake<ClientUniverseMessage<any>>;
+  public messageRxjsSubject = new plugins.smartrx.rxjs.Subject<ClientUniverseMessage<any>>();
   public clientUniverseCache = new ClientUniverseCache();
 
   constructor(optionsArg: IClientOptions) {
@@ -77,7 +77,7 @@ export class ClientUniverse {
   }
 
   public async stop() {
-    await this.smartsocketClient.disconnect();
+    await this.disconnect('triggered');
   }
 
   /**
@@ -85,7 +85,7 @@ export class ClientUniverse {
    * since password validation is done through other means, a connection should always be possible
    */
   public async checkConnection(): Promise<void> {
-    if (!this.smartsocketClient && !this.observableIntake) {
+    if (!this.smartsocketClient) {
       const parsedURL = url.parse(this.options.serverAddress);
       const socketConfig: plugins.smartsocket.ISmartsocketClientOptions = {
         alias: 'universeclient',
@@ -95,7 +95,13 @@ export class ClientUniverse {
         url: parsedURL.protocol + '//' + parsedURL.hostname
       };
       this.smartsocketClient = new SmartsocketClient(socketConfig);
-      this.observableIntake = new plugins.smartrx.ObservableIntake();
+
+      this.smartsocketClient.eventSubject.subscribe(async eventArg => {
+        switch(eventArg) {
+          case 'disconnected':
+            this.disconnect('upstreamEvent');
+        }
+      });
 
       // lets define some basic actions
 
@@ -105,8 +111,14 @@ export class ClientUniverse {
       const socketFunctionUnsubscribe = new plugins.smartsocket.SocketFunction({
         funcName: 'unsubscribe',
         allowedRoles: [],
-        funcDef: async (data: interfaces.IServerUnsubscribeActionPayload) => {
-          throw new Error('TODO');
+        funcDef: async (dataArg: interfaces.IServerUnsubscribeActionPayload) => {
+          const channel = this.clientUniverseCache.channelMap.find(channelArg => {
+            return channelArg.name === dataArg.name;
+          });
+          if (channel) {
+            channel.unsubscribe();
+          }
+          return {};
         }
       });
 
@@ -123,7 +135,7 @@ export class ClientUniverse {
           const clientUniverseMessage = ClientUniverseMessage.createMessageFromMessageDescriptor(
             messageDescriptorArg
           );
-          this.observableIntake.push(clientUniverseMessage);
+          this.messageRxjsSubject.next(clientUniverseMessage);
 
           // lets find the corresponding channel
           const targetChannel = this.getChannel(clientUniverseMessage.targetChannelName);
@@ -151,4 +163,15 @@ export class ClientUniverse {
       });
     }
   }
+
+  public async disconnect(reason: 'upstreamEvent' | 'triggered' = 'triggered', tryReconnect = false) {
+    if ('triggered') {
+      this.smartsocketClient.disconnect();
+    }
+    this.smartsocketClient = null;
+    if (tryReconnect) {
+      await plugins.smartdelay.delayForRandom(5000, 20000);
+      this.checkConnection();
+    }
+  }
 }

ts/smartuniverse.classes.clientuniversechannel.ts

@@ -63,6 +63,10 @@ export class ClientUniverseChannel implements interfaces.IUniverseChannel {
     );
   }
 
+  public unsubscribe() {
+    // TODO: unsubscribe all users
+  }
+
   public async populateSubscriptionToServer() {
     // lets make sure the channel is connected
     if (this.status === 'unsubscribed') {

ts/smartuniverse.classes.universe.ts

@@ -112,6 +112,7 @@ export class Universe {
       funcName: 'subscribeChannel',
       funcDef: async (dataArg, socketConnectionArg) => {
         const universeConnection = new UniverseConnection({
+          universe: this,
           socketConnection: socketConnectionArg,
           authenticationRequests: [dataArg]
         });
@@ -178,7 +179,6 @@ 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.universeconnection.ts

@@ -26,7 +26,7 @@ export class UniverseConnection {
       universeConnection
     );
     universeRef.universeCache.connectionMap.add(universeConnection);
-    console.log('hi')
+    console.log('hi');
   }
 
   /**
@@ -93,6 +93,8 @@ export class UniverseConnection {
     return universeConnection;
   }
 
+  // INSTANCE
+  public universeRef: Universe;
   public terminatedDeferred = plugins.smartpromise.defer();
 
   /**
@@ -100,23 +102,33 @@ export class UniverseConnection {
    */
   public socketConnection: plugins.smartsocket.SocketConnection;
   public authenticationRequests: Array<interfaces.ISocketRequest_SubscribeChannel['request']> = [];
-  public subscribedChannels: UniverseChannel[] = [];
   public authenticatedChannels: UniverseChannel[] = [];
   public failedToJoinChannels: UniverseChannel[] = [];
 
   /**
-   * terminates the connection
+   * disconnect the connection
    */
-  public terminateConnection() {
-    this.socketConnection.socket.disconnect();
+  public async disconnect(reason: 'upstreamevent' | 'triggered' = 'triggered') {
+    if (reason === 'triggered') {
+      await this.socketConnection.disconnect();
+    }
+    this.universeRef.universeCache.connectionMap.remove(this);
     this.terminatedDeferred.resolve();
   }
 
   constructor(optionsArg: {
+    universe: Universe;
     socketConnection: plugins.smartsocket.SocketConnection;
     authenticationRequests: Array<interfaces.ISocketRequest_SubscribeChannel['request']>;
   }) {
     this.authenticationRequests = optionsArg.authenticationRequests;
     this.socketConnection = optionsArg.socketConnection;
+    this.socketConnection.eventSubject.subscribe(async eventArg => {
+      switch (eventArg) {
+        case 'disconnected':
+          await this.disconnect('upstreamevent');
+          break;
+      }
+    });
   }
 }