subscriber added

src/app/pages/home/home-page.component.ts

@@ -6,9 +6,10 @@ import { catchError, exhaustMap, map, tap, take } from 'rxjs/operators';
 import { ProbeService, requesterID } from '../../../commons/service/probe.service';
 
 import { Interface } from '@overflow/model/net/nic';
-import { Zone, DiscoverHost } from '@overflow/model/discovery';
+import { Zone, DiscoverHost, Host, Port, Service } from '@overflow/model/discovery';
 import { toMetaIPType, MetaIPTypeEnum } from '@overflow/model/meta';
 import { Address } from '../../../commons/component/nic-dropdown.component';
+import { RPCSubscriber } from '@overflow/commons/ui/decorator/RPCSubscriber';
 
 
 @Component({
@@ -25,7 +26,9 @@ export class HomePageComponent implements OnInit {
 
   constructor(
     private probeService: ProbeService,
-  ) { }
+  ) {
+    this.probeService.subscribeNotification(this);
+  }
 
   ngOnInit() {
   }
@@ -38,7 +41,7 @@ export class HomePageComponent implements OnInit {
     }, 3000);
   }
 
-  discoverHost() {
+  discoverHost(dfd: string) {
     const zone: Zone = {
       network: '192.168.1.0/24',
       iface: 'enp3s0',
@@ -86,4 +89,75 @@ export class HomePageComponent implements OnInit {
     // ).subscribe();
   }
 
+  /**
+   * DiscoverHost
+   */
+  @RPCSubscriber({ method: 'DiscoveryService.Queueing' })
+  public DiscoveryQueueing(queueingDate: Date) {
+    console.log('DiscoveryQueueing', queueingDate);
+  }
+
+  /**
+   * DiscoverHost
+   */
+  @RPCSubscriber({ method: 'DiscoveryService.QueueingFailed' })
+  public DiscoveryQueueingFailed(queueingFailedDate: Date) {
+    console.log('DiscoveryQueueingFailed', queueingFailedDate);
+  }
+
+  /**
+   * DiscoverHost
+   */
+  @RPCSubscriber({ method: 'DiscoveryService.QueueingTimeout' })
+  public DiscoveryQueueingTimeout(queueingTimeoutDate: Date) {
+    console.log('DiscoveryQueueingTimeout', queueingTimeoutDate);
+  }
+
+  /**
+   * DiscoverHost
+   */
+  @RPCSubscriber({ method: 'DiscoveryService.DiscoveryStart' })
+  public DiscoveryStart(startDate: Date) {
+    console.log('DiscoveryStart', startDate);
+  }
+
+  /**
+   * DiscoverHost
+   */
+  @RPCSubscriber({ method: 'DiscoveryService.DiscoveryStop' })
+  public DiscoveryStop(elapsed: number) {
+    console.log('DiscoveryStop', elapsed);
+  }
+
+  /**
+   * DiscoverHost
+   */
+  @RPCSubscriber({ method: 'DiscoveryService.DiscoveryError' })
+  public DiscoveryError(err: Error) {
+    console.log('DiscoveryError', err);
+  }
+
+  /**
+   * DiscoverHost
+   */
+  @RPCSubscriber({ method: 'DiscoveryService.DiscoveredHost' })
+  public DiscoveredHost(host: Host) {
+    console.log('DiscoveredHost', host);
+  }
+
+  /**
+   * DiscoverHost
+   */
+  @RPCSubscriber({ method: 'DiscoveryService.DiscoveredPort' })
+  public DiscoveredPort(port: Port) {
+    console.log('DiscoveredPort', port);
+  }
+
+  /**
+   * DiscoverHost
+   */
+  @RPCSubscriber({ method: 'DiscoveryService.DiscoveredService' })
+  public DiscoveredService(service: Service) {
+    console.log('DiscoveredService', service);
+  }
 }

src/commons/service/probe.service.ts

@@ -7,9 +7,35 @@ import {
   JSONClientCodec
 } from '@overflow/rpc-js';
 import { Subscription } from 'rxjs';
+import { TypeUtil, Class, Annotation, PropertyKeyType, Method } from '@overflow/core-js';
+import { RPCSubscriberDecoratorAttribute } from '@overflow/commons/ui/decorator/RPCSubscriber';
 
 export const requesterID = 'scannerUser';
 
+export interface SubscriberMethod {
+  className: PropertyKeyType;
+  methodName: PropertyKeyType;
+  parameterTypes: string[];
+
+  method: Method;
+  instance: any;
+}
+
+export class SubscriberParameterError extends Error {
+  public constructor(message?: string) {
+    super(message);
+    Object.setPrototypeOf(this, new.target.prototype);
+  }
+}
+
+export class SubscriberExistError extends Error {
+  public constructor(message?: string) {
+    super(message);
+    Object.setPrototypeOf(this, new.target.prototype);
+  }
+}
+
+
 const config: WebSocketClientRWCConfig = {
   url: 'ws://localhost:60000/scanner?requesterID=' + requesterID
 };
@@ -22,21 +48,35 @@ const codec = new JSONClientCodec();
 export class ProbeService extends Client {
   private notiSubscription: Subscription;
 
+  private subscriberMap: Map<any, string[]>;
+  private subscriberMethodMap: Map<string, SubscriberMethod[]>;
+
   public constructor() {
     super(codec, rwc);
 
+    this.subscriberMap = new Map();
+    this.subscriberMethodMap = new Map();
+
     this.connect();
 
     this.notiSubscription = this.notification()
       .subscribe((notiCodec: ClientNotificationCodec) => {
-        console.log(notiCodec.method());
+        const method = notiCodec.method();
         const params = notiCodec.params();
-        if (undefined !== params) {
-          params.forEach(param => {
 
-            console.log(param);
-          });
+        const subscriberMethods: SubscriberMethod[] = this.subscriberMethodMap.get(method);
+        if (undefined === subscriberMethods) {
+          console.warn(`Subscriber for method[${method}] is not exist`);
+          return;
         }
+        subscriberMethods.forEach((subscriberMethod) => {
+          try {
+            const args = this.converParams(params, subscriberMethod.parameterTypes);
+            subscriberMethod.method.invoke(subscriberMethod.instance, ...args);
+          } catch (error) {
+            console.error(error);
+          }
+        });
       });
     this.connectionStatus().subscribe(isConnected => {
       if (!isConnected) {
@@ -45,5 +85,136 @@ export class ProbeService extends Client {
         console.log('connected');
       }
     });
+
   }
+
+  /**
+   * subscribeNotification
+   */
+  public subscribeNotification(subscriber: any) {
+    if (this.subscriberMap.has(subscriber)) {
+      throw new SubscriberExistError();
+    }
+
+    const type = TypeUtil.getType(subscriber);
+
+    const clazz = Class.forType(type);
+    if (undefined === clazz) {
+      console.log(`Type[${subscriber.name}] is not decorated type`);
+      return;
+    }
+
+    const subscriberMethodNames: string[] = [];
+    this.subscriberMap.set(subscriber, subscriberMethodNames);
+
+    const methods = clazz.getMethods();
+    methods.forEach((method, propertyKey) => {
+      const annon = method.getAnnotation<Annotation<RPCSubscriberDecoratorAttribute>>(Annotation);
+      if (undefined === annon) {
+        return;
+      }
+
+      const subscriberMethodName = annon.attribute.method;
+      subscriberMethodNames.push(subscriberMethodName);
+      let subscriberMethods: SubscriberMethod[] = this.subscriberMethodMap.get(subscriberMethodName);
+      if (undefined === subscriberMethods) {
+        subscriberMethods = [];
+        this.subscriberMethodMap.set(subscriberMethodName, subscriberMethods);
+      }
+
+      const paramTypes = this.getParamTypes(method);
+
+      const subscriberMethod: SubscriberMethod = {
+        className: clazz.getName(),
+        methodName: method.getName(),
+        parameterTypes: paramTypes,
+        method: method,
+        instance: subscriber,
+      };
+
+      subscriberMethods.push(subscriberMethod);
+    });
+  }
+
+  /**
+   * subscribeNotification
+   */
+  public unsubscribeNotification(subscriber: any) {
+    if (!this.subscriberMap.has(subscriber)) {
+      return;
+    }
+
+    const subscriberMethodNames = this.subscriberMap.get(subscriber);
+    this.subscriberMap.delete(subscriber);
+    subscriberMethodNames.forEach(methodName => {
+      if (!this.subscriberMethodMap.has(methodName)) {
+        return;
+      }
+      const subscriberMethods = this.subscriberMethodMap.get(methodName);
+      subscriberMethods.some((subscriberMethod, index): boolean => {
+        if (subscriberMethod.instance === subscriber) {
+          delete subscriberMethods[index];
+        }
+        return false;
+      });
+    });
+  }
+
+  private getParamTypes(method: Method): string[] {
+    if (undefined === method || null === method || 0 === method.getParameterCount()) {
+      return [];
+    }
+
+    const parameters = method.getParameters();
+    const results: string[] = [];
+    for (let indexI = 0; indexI < parameters.length; indexI++) {
+      const paramType = parameters[indexI].getType();
+      results.push(paramType.name);
+    }
+    return results;
+  }
+
+  private converParams(params: string[], paramTypes: string[]): any[] {
+    const results: any[] = [];
+
+    if (undefined === params || null === params || 0 === params.length) {
+      return results;
+    }
+    if (undefined === paramTypes || null === paramTypes || 0 === paramTypes.length) {
+      return results;
+    }
+    if (params.length !== paramTypes.length) {
+      throw new SubscriberParameterError(`Count is not same from server[${params.length}] and method[${paramTypes.length}]`);
+    }
+    for (let indexI = 0; indexI < params.length; indexI++) {
+      const param = params[indexI];
+      const type = paramTypes[indexI];
+      switch (type) {
+        case 'Object':
+        case 'Array':
+        case 'Map':
+          results.push(JSON.parse(param));
+          break;
+        case 'String':
+          results.push(param);
+          break;
+        case 'Number':
+          results.push(Number(param));
+          break;
+        case 'Boolean':
+          results.push(Boolean(param));
+          break;
+        case 'Date':
+          results.push(new Date(Number(param)));
+          break;
+        case 'Function':
+          throw new SubscriberParameterError(`Function type [${indexI}] is not allowed`);
+        default:
+          throw new SubscriberParameterError(`${type} type parameter[${indexI}] is not allowed`);
+      }
+    }
+    return results;
+  }
+
+
 }