1.0.86

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@pushrocks/smartuniverse",
-  "version": "1.0.77",
+  "version": "1.0.86",
   "lockfileVersion": 1,
   "requires": true,
   "dependencies": {
@@ -1243,7 +1243,7 @@
     },
     "esutils": {
       "version": "2.0.3",
-      "resolved": "https://verdaccio.lossless.one/esutils/-/esutils-2.0.3.tgz",
+      "resolved": "https://registry.npmjs.org/esutils/-/esutils-2.0.3.tgz",
       "integrity": "sha512-kVscqXk4OCp68SZ0dkgEKVi6/8ij300KBWTJq32P/dYeWTSwK41WyTxalN1eRmA5Z9UU/LX9D7FWSmV9SAYx6g==",
       "dev": true
     },
@@ -1949,7 +1949,7 @@
     },
     "minimist": {
       "version": "0.0.8",
-      "resolved": "https://verdaccio.lossless.one/minimist/-/minimist-0.0.8.tgz",
+      "resolved": "https://registry.npmjs.org/minimist/-/minimist-0.0.8.tgz",
       "integrity": "sha1-hX/Kv8M5fSYluCKCYuhqp6ARsF0=",
       "dev": true
     },
@@ -2364,7 +2364,7 @@
     },
     "resolve": {
       "version": "1.12.0",
-      "resolved": "https://verdaccio.lossless.one/resolve/-/resolve-1.12.0.tgz",
+      "resolved": "https://registry.npmjs.org/resolve/-/resolve-1.12.0.tgz",
       "integrity": "sha512-B/dOmuoAik5bKcD6s6nXDCjzUKnaDvdkRyAk6rsmsKLipWj4797iothd7jmmUhWTfinVMU+wc56rYKsit2Qy4w==",
       "dev": true,
       "requires": {

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@pushrocks/smartuniverse",
-  "version": "1.0.77",
+  "version": "1.0.86",
   "private": false,
   "description": "messaging service for your micro services",
   "main": "dist/index.js",
@@ -25,6 +25,7 @@
     "rxjs": "*"
   },
   "dependencies": {
+    "@apiglobal/typedrequest-interfaces": "^1.0.7",
     "@pushrocks/lik": "^3.0.11",
     "@pushrocks/smartdelay": "^2.0.3",
     "@pushrocks/smartexpress": "^3.0.40",

test/test.ts

@@ -69,9 +69,53 @@ tap.test('a second client should be able to subscibe', async () => {
   });
 
   testClientUniverse2.addChannel(testChannelData.channelName, testChannelData.channelPass);
+  await testClientUniverse2.start();
 });
 
-tap.test('should receive a message correctly', async () => {});
+tap.test('should receive a message correctly', async (tools) => {
+  const done = tools.defer();
+  const testChannel = testClientUniverse.getChannel(testChannelData.channelName);
+  const testChannel2 = testClientUniverse2.getChannel(testChannelData.channelName);
+  const subscription = testChannel2.subscribe(messageArg => {
+    console.log('Yay##########');
+    done.resolve();
+  });
+  await testChannel.sendMessage({
+    messageText: 'hellothere'
+  });
+  await done.promise;
+});
+
+interface IDemoReqRes {
+  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 disconnect the client correctly', async () => {
   await testClientUniverse.stop();

ts/index.ts

@@ -9,4 +9,8 @@ export * from './smartuniverse.classes.universecache';
 export * from './smartuniverse.classes.universechannel';
 export * from './smartuniverse.classes.universemessage';
 
+// Reaction Response
+export * from './smartuniverse.classes.reactionrequest';
+export * from './smartuniverse.classes.reactionresponse';
+
 export * from './interfaces';

ts/interfaces/socketfunctionrequests.ts

@@ -7,7 +7,7 @@ export interface ISocketRequest_SubscribeChannel {
     passphrase: string;
   };
   response: {
-    subscriptionStatus: 'subscribed' | 'unsubscribed'
+    subscriptionStatus: 'subscribed' | 'unsubscribed';
   };
 }
 
@@ -15,6 +15,6 @@ export interface ISocketRequest_ProcessMessage {
   method: 'processMessage';
   request: interfaces.IUniverseMessage;
   response: {
-    messageStatus: 'ok' | 'channel not found'
+    messageStatus: 'ok' | 'channel not found';
   };
-}
+}

ts/interfaces/universeactions.interfaces.ts

@@ -1,10 +1,3 @@
-export type IServerCallActions =
-  | 'channelSubscription'
-  | 'processMessage'
-  | 'channelUnsubscribe'
-  | 'terminateConnection';
-
-
 export interface IServerUnsubscribeActionPayload {
   name: string;
 }

ts/interfaces/universemessage.interfaces.ts

@@ -1,7 +1,6 @@
 export interface IMessageCreator {
   messageText: string;
   payload?: string | number | any;
-  payloadStringType?: 'Buffer' | 'string' | 'object';
 }
 
 /**

ts/smartuniverse.classes.clientuniverse.ts

@@ -21,7 +21,7 @@ export interface IClientOptions {
 export class ClientUniverse {
   public options;
   public smartsocketClient: plugins.smartsocket.SmartsocketClient;
-  public observableIntake: plugins.smartrx.ObservableIntake<ClientUniverseMessage>;
+  public observableIntake: plugins.smartrx.ObservableIntake<ClientUniverseMessage<any>>;
   public clientUniverseCache = new ClientUniverseCache();
 
   constructor(optionsArg: IClientOptions) {
@@ -147,7 +147,7 @@ export class ClientUniverse {
       await this.smartsocketClient.connect();
       plugins.smartlog.defaultLogger.log('info', 'universe client connected successfully');
       await this.clientUniverseCache.channelMap.forEach(async clientUniverseChannelArg => {
-        await clientUniverseChannelArg.subscribe();
+        await clientUniverseChannelArg.populateSubscriptionToServer();
       });
     }
   }

ts/smartuniverse.classes.clientuniversechannel.ts

@@ -38,7 +38,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<ClientUniverseMessage>();
+  private subject = new plugins.smartrx.rxjs.Subject<ClientUniverseMessage<any>>();
 
   // refs
   public clientUniverseRef: ClientUniverse;
@@ -53,34 +53,33 @@ export class ClientUniverseChannel implements interfaces.IUniverseChannel {
    * subscribes to a channel
    * tells the universe about this instances interest into a channel
    */
-  public async subscribe(observerArg?: plugins.smartrx.rxjs.Observer<any>) {
+  public subscribe(observingFunctionArg: (messageArg: ClientUniverseMessage<any>) => void) {
+
+    return this.subject.subscribe(
+      messageArg => {
+        observingFunctionArg(messageArg);
+      },
+      error => console.log(error)
+    );
+  }
+
+  public async populateSubscriptionToServer() {
     // lets make sure the channel is connected
     if (this.status === 'unsubscribed') {
-      const response = await this.clientUniverseRef.smartsocketClient.serverCall<interfaces.ISocketRequest_SubscribeChannel>(
-        'subscribeChannel',
-        {
-          name: this.name,
-          passphrase: this.passphrase
-        }
-      );
+      const response = await this.clientUniverseRef.smartsocketClient.serverCall<
+        interfaces.ISocketRequest_SubscribeChannel
+      >('subscribeChannel', {
+        name: this.name,
+        passphrase: this.passphrase
+      });
       this.status = response.subscriptionStatus;
     }
-
-    if (observerArg) {
-      return this.subject.subscribe(observerArg);
-    }
-
   }
 
-  public async emitMessageLocally(messageArg: ClientUniverseMessage) {
+  public async emitMessageLocally(messageArg: ClientUniverseMessage<any>) {
     this.subject.next(messageArg);
   }
 
-  public askForReaction(reactionRequest: ReactionRequest): ReactionResponse {
-    const reactionResponse = new ReactionResponse();
-    return reactionResponse;
-  }
-
   /**
    * sends a message towards the server
    * @param messageArg
@@ -93,8 +92,7 @@ export class ClientUniverseChannel implements interfaces.IUniverseChannel {
       passphrase: this.passphrase,
       targetChannelName: this.name,
       messageText: messageArg.messageText,
-      payload: messageArg.payload,
-      payloadStringType: messageArg.payloadStringType
+      payload: messageArg.payload
     };
     await this.clientUniverseRef.smartsocketClient.serverCall(
       'processMessage',

ts/smartuniverse.classes.clientuniversemessage.ts

@@ -2,7 +2,7 @@ import * as plugins from './smartuniverse.plugins';
 
 import * as interfaces from './interfaces';
 
-export class ClientUniverseMessage implements interfaces.IUniverseMessage {
+export class ClientUniverseMessage<T> implements interfaces.IUniverseMessage {
   // ======
   // STATIC
   // ======
@@ -22,8 +22,7 @@ export class ClientUniverseMessage implements interfaces.IUniverseMessage {
   public smartTimestamp: plugins.smarttime.TimeStamp;
   public messageText: string;
   public passphrase: string;
-  public payload: any;
-  public payloadStringType;
+  public payload: T;
   public targetChannelName: string;
 
   constructor(messageArg: interfaces.IUniverseMessage) {

ts/smartuniverse.classes.reactionrequest.ts

@@ -1,5 +1,66 @@
 import * as plugins from './smartuniverse.plugins';
+import { UniverseChannel } from './smartuniverse.classes.universechannel';
+import { ClientUniverseChannel } from './smartuniverse.classes.clientuniversechannel';
+import { ReactionResult } from './smartuniverse.classes.reactionresult';
+import { UniverseMessage } from './smartuniverse.classes.universemessage';
+import { ClientUniverseMessage } from './smartuniverse.classes.clientuniversemessage';
 
-export class ReactionRequest {
-  
+export interface IReactionRequestConstructorOptions<T extends plugins.typedrequestInterfaces.ITypedRequest> {
+  method: T['method'];
+}
+
+export interface ICombinatorPayload<T extends plugins.typedrequestInterfaces.ITypedRequest> {
+  /**
+   * needed for tying responses to requests
+   */
+  id: string;
+  typedRequestPayload: {
+    method: T['method'],
+    request : T['request'],
+    response: T['response']
+  };
+}
+
+export class ReactionRequest<T extends plugins.typedrequestInterfaces.ITypedRequest> {
+  public method: T['method'];
+
+  constructor(optionsArg: IReactionRequestConstructorOptions<T>) {
+    this.method = optionsArg.method;
+  }
+
+  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();
+    for (const channel of channelsArg) {
+      subscriptionMap.add(channel.subscribe((messageArg: UniverseMessage<ICombinatorPayload<T>> | ClientUniverseMessage<ICombinatorPayload<T>>) => {
+        if (messageArg.messageText === 'reactionResponse' && messageArg.payload.typedRequestPayload.method === this.method) {
+          const payload: ICombinatorPayload<T> = messageArg.payload;
+          if (payload.id !== requestId) {
+            return;
+          }
+          reactionResult.pushReactionResponse(payload.typedRequestPayload.response);
+        }
+      }));
+      const payload: ICombinatorPayload<T> = {
+        id: requestId,
+        typedRequestPayload: {
+          method: this.method,
+          request: requestDataArg,
+          response: null
+        }
+      };
+      channel.sendMessage({
+        messageText: 'reactionRequest',
+        payload
+      });
+    }
+    plugins.smartdelay.delayFor(timeoutMillisArg).then(async () => {
+      await subscriptionMap.forEach(subscriptionArg => {
+        subscriptionArg.unsubscribe();
+      });
+      reactionResult.complete();
+    });
+    return reactionResult;
+  }
 }

ts/smartuniverse.classes.reactionresponse.ts

@@ -1,3 +1,59 @@
 import * as plugins from './smartuniverse.plugins';
 
-export class ReactionResponse {} 
+import { ICombinatorPayload } from './smartuniverse.classes.reactionrequest';
+import { UniverseChannel } from './smartuniverse.classes.universechannel';
+import { ClientUniverseChannel } from './smartuniverse.classes.clientuniversechannel';
+import { UniverseMessage } from './smartuniverse.classes.universemessage';
+import { ClientUniverseMessage } from './smartuniverse.classes.clientuniversemessage';
+
+export type TReactionResponseFuncDef<T extends plugins.typedrequestInterfaces.ITypedRequest> = (dataArg: T['request']) => Promise<T['response']>;
+
+export interface IReactionResponseConstructorOptions<
+  T extends plugins.typedrequestInterfaces.ITypedRequest
+> {
+  method: T['method'];
+  channels: Array<UniverseChannel | ClientUniverseChannel>;
+  funcDef: TReactionResponseFuncDef<T>;
+}
+
+export class ReactionResponse<T extends plugins.typedrequestInterfaces.ITypedRequest> {
+  public method: T['method'];
+  public channels = new plugins.lik.Objectmap<UniverseChannel | ClientUniverseChannel>();
+  public funcDef: TReactionResponseFuncDef<T>;
+
+  constructor(optionsArg: IReactionResponseConstructorOptions<T>) {
+    this.method = optionsArg.method;
+    this.channels.addArray(optionsArg.channels);
+    this.funcDef = optionsArg.funcDef;
+    for (const channel of this.channels.getArray()) {
+      channel.subscribe(messageArg => {
+        this.processMessageForReaction(channel, messageArg);
+      });
+    }
+  }
+
+  private async processMessageForReaction(
+    channelArg: UniverseChannel | ClientUniverseChannel,
+    messageArg:
+      | UniverseMessage<ICombinatorPayload<T>>
+      | ClientUniverseMessage<ICombinatorPayload<T>>
+  ) {
+    if (
+      messageArg.messageText === 'reactionRequest' &&
+      messageArg.payload.typedRequestPayload.method === this.method
+    ) {
+      const response: T['response'] = await this.funcDef(messageArg.payload.typedRequestPayload.request);
+      const payload: ICombinatorPayload<T> = {
+        ...messageArg.payload,
+        typedRequestPayload: {
+          ...messageArg.payload.typedRequestPayload,
+          response
+        }
+      };
+      channelArg.sendMessage({
+        messageText: 'reactionResponse',
+        payload 
+      });
+    }
+  }
+}

ts/smartuniverse.classes.reactionresult.ts

@@ -0,0 +1,52 @@
+import * as plugins from './smartuniverse.plugins';
+import { ReactionResponse } from './smartuniverse.classes.reactionresponse';
+
+export class ReactionResult<T extends plugins.typedrequestInterfaces.ITypedRequest> {
+  private resultReplaySubject = new plugins.smartrx.rxjs.ReplaySubject<T['response']>();
+  private endResult: Array<T['response']> = [];
+  private completeDeferred = plugins.smartpromise.defer<Array<T['response']>>();
+
+  constructor () {
+    this.resultSubscribe(responseArg => {
+      this.endResult.push(responseArg);
+    });
+  }
+
+  public resultSubscribe(observerArg: (responseArg: T['response']) => void) {
+    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.resultReplaySubject.next(responseArg);
+  }
+  
+  /**
+   * completes the ReactionResult
+   */
+  public async complete() {
+    this.completeDeferred.resolve(this.endResult);
+  }
+}

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;
     });
@@ -105,13 +105,12 @@ export class Universe {
     // add the role to smartsocket
     this.smartsocket.addSocketRoles([ClientRole]);
 
-    const socketFunctionSubscription = new plugins.smartsocket.SocketFunction<interfaces.ISocketRequest_SubscribeChannel>({
+    const socketFunctionSubscription = new plugins.smartsocket.SocketFunction<
+      interfaces.ISocketRequest_SubscribeChannel
+    >({
       allowedRoles: [ClientRole], // there is only one client role, Authentication happens on another level
       funcName: 'subscribeChannel',
-      funcDef: async (
-        dataArg,
-        socketConnectionArg
-      ) => {
+      funcDef: async (dataArg, socketConnectionArg) => {
         const universeConnection = new UniverseConnection({
           socketConnection: socketConnectionArg,
           authenticationRequests: [dataArg]

ts/smartuniverse.classes.universecache.ts

@@ -24,7 +24,7 @@ export class UniverseCache {
   /**
    * stores messages for this instance
    */
-  public messageMap = new Objectmap<UniverseMessage>();
+  public messageMap = new Objectmap<UniverseMessage<any>>();
 
   /**
    * stores the channels that are available within the universe
@@ -54,7 +54,7 @@ export class UniverseCache {
    * @param messageArg
    * @param attachedPayloadArg
    */
-  public async addMessage(messageArg: UniverseMessage) {
+  public async addMessage(messageArg: UniverseMessage<any>) {
     messageArg.setUniverseCache(this);
     UniverseChannel.authorizeAMessageForAChannel(this, messageArg);
     this.messageMap.add(messageArg);
@@ -69,7 +69,7 @@ export class UniverseCache {
   public readMessagesYoungerThan(
     unixTimeArg?: number,
     channelName?: string
-  ): Observable<UniverseMessage> {
+  ): Observable<UniverseMessage<any>> {
     const messageObservable = from(this.messageMap.getArray()).pipe(
       filter(messageArg => {
         return messageArg.smartTimestamp.isYoungerThanMilliSeconds(this.destructionTime);

ts/smartuniverse.classes.universechannel.ts

@@ -52,7 +52,7 @@ export class UniverseChannel {
    */
   public static authorizeAMessageForAChannel(
     universeCacheArg: UniverseCache,
-    universeMessageArg: UniverseMessage
+    universeMessageArg: UniverseMessage<any>
   ): UniverseChannel {
     const foundChannel = universeCacheArg.channelMap.find(universeChannel => {
       const result = universeChannel.authenticate(universeMessageArg);
@@ -85,7 +85,7 @@ export class UniverseChannel {
    */
   public name: string;
   public universeRef: Universe;
-  private subject = new plugins.smartrx.rxjs.Subject<UniverseMessage>();
+  private subject = new plugins.smartrx.rxjs.Subject<UniverseMessage<any>>();
 
   /**
    * the passphrase for the channel
@@ -103,7 +103,7 @@ export class UniverseChannel {
    * # the messages channelName against the unverseChannel's name
    * # the messages password against the universeChannel's password
    */
-  public authenticate(universeMessageArg: UniverseMessage): boolean {
+  public authenticate(universeMessageArg: UniverseMessage<any>): boolean {
     return (
       this.name === universeMessageArg.targetChannelName &&
       this.passphrase === universeMessageArg.passphrase
@@ -114,10 +114,10 @@ export class UniverseChannel {
    * pushes a message to clients
    * @param messageArg
    */
-  public async push(messageArg: UniverseMessage) {
+  public async push(messageArg: UniverseMessage<any>) {
     this.subject.next(messageArg);
     const universeConnectionsWithChannelAccess: UniverseConnection[] = [];
-    this.universeRef.universeCache.connectionMap.forEach(async socketConnection => {
+    await this.universeRef.universeCache.connectionMap.forEach(async socketConnection => {
       if (socketConnection.authenticatedChannels.includes(this)) {
         universeConnectionsWithChannelAccess.push(socketConnection);
       }
@@ -131,8 +131,7 @@ export class UniverseChannel {
         passphrase: messageArg.passphrase,
         targetChannelName: this.name,
         messageText: messageArg.messageText,
-        payload: messageArg.payload,
-        payloadStringType: messageArg.payloadStringType
+        payload: messageArg.payload
       };
       smartsocket.clientCall(
         'processMessage',
@@ -143,8 +142,13 @@ export class UniverseChannel {
   }
 
   // functions to interact with a channel locally
-  public async subscribe(observer: plugins.smartrx.rxjs.Observer<any>) {
-    return this.subject.subscribe(observer);
+  public subscribe(observingFunctionArg: (messageArg: UniverseMessage<any>) => void) {
+    return this.subject.subscribe(
+      messageArg => {
+        observingFunctionArg(messageArg);
+      },
+      error => console.log(error)
+    );
   }
 
   /**
@@ -155,7 +159,6 @@ export class UniverseChannel {
       id: plugins.smartunique.shortId(),
       messageText: messageDescriptor.messageText,
       payload: messageDescriptor.payload,
-      payloadStringType: messageDescriptor.payloadStringType,
       targetChannelName: this.name,
       passphrase: this.passphrase,
       timestamp: Date.now()

ts/smartuniverse.classes.universeconnection.ts

@@ -26,6 +26,7 @@ export class UniverseConnection {
       universeConnection
     );
     universeRef.universeCache.connectionMap.add(universeConnection);
+    console.log('hi')
   }
 
   /**

ts/smartuniverse.classes.universemessage.ts

@@ -13,7 +13,7 @@ import { SocketConnection } from '@pushrocks/smartsocket';
  * represents a message within a universe
  * acts as a container to save message states like authentication status
  */
-export class UniverseMessage implements interfaces.IUniverseMessage {
+export class UniverseMessage<T> implements interfaces.IUniverseMessage {
   public static createMessageFromPayload(
     socketConnectionArg: SocketConnection,
     dataArg: interfaces.IUniverseMessage
@@ -28,8 +28,7 @@ export class UniverseMessage implements interfaces.IUniverseMessage {
   public smartTimestamp: TimeStamp;
   public messageText: string;
   public passphrase: string;
-  public payload: any;
-  public payloadStringType;
+  public payload: T;
   public targetChannelName: string;
   public socketConnection: SocketConnection;
 

ts/smartuniverse.plugins.ts

@@ -3,6 +3,13 @@ import * as path from 'path';
 
 export { path };
 
+// apiglobal scope
+import * as typedrequestInterfaces from '@apiglobal/typedrequest-interfaces';
+
+export {
+  typedrequestInterfaces
+};
+
 // pushrocks scope
 import * as lik from '@pushrocks/lik';
 import * as smarthash from '@pushrocks/smarthash';