mkarkowski/nope
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

adding event dispatching

Browse Source
This commit is contained in:
Martin Karkowski 2020-09-17 08:05:45 +02:00
parent a7e7b22bcf
commit 4b4284554b
5 changed files with 522 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
lib/communication/eventLayer.ts
View File

@ -1,5 +1,6 @@
import { EventEmitter } from "events";
import { ICommunicationInterface, requestTaskMsg, responseTaskMsg, availableServices } from "../dispatcher/nopeDispatcher";
import { Logger } from "winston";
import { ICommunicationInterface, requestTaskMsg, responseTaskMsg, availableServices, availableTopics, externalEvent } from "../dispatcher/nopeDispatcher";
/**
* A Communication Layer for the Dispatchers.
@ -43,6 +44,28 @@ export class EventLayer implements ICommunicationInterface {
){
}
onBonjour(cb: (dispatcher: string) => void): void {
this._emitter.on('bonjour', cb);
}
emitBonjour(dispatcher: string): void {
this._emitter.emit('bonjour', dispatcher);
}
emitNewTopicsAvailable(topics: availableTopics): void {
this._emitter.emit('topics',topics)
}
onNewTopicsAvailable(cb: (topics: availableTopics) => void) {
this._emitter.on('topics',cb)
}
onEvent(event: string, cb: (data: externalEvent) => void): void {
this._emitter.on('event_'+event, cb);
}
emitEvent(event: string, data: externalEvent): void {
this._emitter.emit('event_'+event, data)
}
offEvent(event: string, cb: (data: externalEvent) => void): void {
this._emitter.off('event_'+event, cb);
}
protected _emitter = new EventEmitter();
}

4
lib/dispatcher/getDispatcher.ts
View File

@ -1,11 +1,11 @@
import { getSingleton } from "../helpers/singletonMethod";
import { nopeDispatcher, nopeDispatcherOptions } from "./nopeDispatcher";
import { nopeDispatcher, nopeRpcDispatcherOptions } from "./nopeDispatcher";
/**
* Function to extract a Singleton Dispatcher
* @param options The provided options for the Dispatcher
*/
export function getDispatcher(options: nopeDispatcherOptions) {
export function getRpcDispatcher(options: nopeRpcDispatcherOptions) {
const container = getSingleton('nopeBackendDispatcher.instance', () => {
return new nopeDispatcher(options);

8
lib/dispatcher/getLinkedDispatcher.ts
View File

@ -8,15 +8,15 @@
import { getSingleton } from "../helpers/singletonMethod";
import { nopeObservable } from "../observables/nopeObservable";
import { getDispatcher } from "./getDispatcher";
import { nopeDispatcherOptions } from './nopeDispatcher';
import { getRpcDispatcher } from "./getDispatcher";
import { nopeRpcDispatcherOptions } from './nopeDispatcher';
import { IExportFunctionToDispatcherParameters, IExportMethodToDispatcherParameters, IExportPropertyToDispatcherParameters } from "./nopeDispatcherDecorators";
/**
* Returns a Dispatcher.
* @param options
*/
export function getLinkedDispatcher(options: nopeDispatcherOptions) {
export function getLinkedDispatcher(options: nopeRpcDispatcherOptions) {
const container = getSingleton('nopeBackendDispatcher.container', () => {
return {
methods: new Map<string, {
@ -37,7 +37,7 @@ export function getLinkedDispatcher(options: nopeDispatcherOptions) {
}
});
const dispatcher = getDispatcher(options);
const dispatcher = getRpcDispatcher(options);
// Iterate over the Methods
for (const [uri, settings] of container.instance.methods.entries()) {

496
lib/dispatcher/nopeDispatcher.ts
View File

@ -1,5 +1,7 @@
import { generateId } from '../helpers/idMethods';
import { Logger } from "winston";
import { nopeObservable, pipe, observableCallback } from '../observables/nopeObservable';
import { observable } from 'rxjs';
/**
* A Layer to communicate.
@ -84,6 +86,96 @@ export interface ICommunicationInterface {
* @memberof ICommunicationInterface
*/
onNewServicesAvailable(cb: (services: availableServices) => void);
/**
* Function to Emit new Services. They will then be shared to other sytems.
*
* @param {availableTopics} topics
* @memberof ICommunicationInterface
*/
emitNewTopicsAvailable(topics: availableTopics): void;
/**
* Function to register a new Callback, which will be called if new Services are available.
*
* @param {(topics: availableTopics) => void} cb The Callback to Call.
* @memberof ICommunicationInterface
*/
onNewTopicsAvailable(cb: (topics: availableTopics) => void);
/**
* Function to subscribe to an event
*
* @param {string} event The Event name (Usually the Topic.)
* @param {(data: externalEvent) => void} cb The Callback which should be used to call if there are new Events.
* @memberof ICommunicationInterface
*/
onEvent(event: string, cb: (data: externalEvent) => void): void;
/**
* Function used to emit an event on the given event channel
*
* @param {string} event The Name of the Event
* @param {externalEvent} data A datapacket describing the Event
* @memberof ICommunicationInterface
*/
emitEvent(event: string, data: externalEvent): void
/**
* Function to unregister an event listener
*
* @param {string} event The Name of the Event
* @param {(data: externalEvent) => void} cb The desired Callback
* @memberof ICommunicationInterface
*/
offEvent(event: string, cb: (data: externalEvent) => void): void;
/**
* Function, to subscribe to "Bonjour"-Messages of the Dispatcher Dispatchers.
*
* @param {(dispatcher: string) => void} cb The callback which should be called.
* @memberof ICommunicationInterface
*/
onBonjour(cb: (dispatcher: string) => void): void;
/**
* Function to emit a Bonjour message.
*
* @param {string} dispatcher The ID of the new Dispatcher.
* @memberof ICommunicationInterface
*/
emitBonjour(dispatcher: string): void;
}
export type availableTopics = {
/**
* The Id of the Dispatcher
*
* @type {string}
*/
dispatcher: string,
/**
* The List of published Topics (only the registered ones).
*
* @type {string[]}
*/
published: string[]
/**
* The List of subscribed Topics (only the registered ones).
*
* @type {string[]}
*/
subscribed: string[]
}
export type externalEvent = {
type: 'event'
data: any,
topic: string,
sender: string,
timestamp?: number
}
export type requestOfService = {
@ -149,12 +241,6 @@ export type requestTaskMsg = {
*/
resultSink: string
// /**
// * Flag, indicating,
// *
// * @type {boolean}
// */
// deleteAfterCalling: boolean;
}
export type callOptions = {
@ -223,7 +309,7 @@ export type availableServices = {
services: string[]
}
export type nopeDispatcherOptions = {
export type nopeRpcDispatcherOptions = {
communicator: ICommunicationInterface
}
@ -262,6 +348,11 @@ export class nopeDispatcher {
public methodInterfaceWithOptions: { [index: string]: <T>(optins: callOptions, ...args) => Promise<T> }
public methodInterface: { [index: string]: <T>(...args) => Promise<T> }
protected _mappingOfRemoteDispatchersAndTopics: Map<string, availableTopics>;
protected _externalSubscribed: Set<string>;
protected _externalPublished: Set<string>;
/**
* Internal Element to store the running tasks.
*
@ -281,7 +372,7 @@ export class nopeDispatcher {
* @param {ICommunicationInterface} communicator The Communication Layer which should be used.
* @memberof nopeDispatcher
*/
constructor(public options: nopeDispatcherOptions) {
constructor(public options: nopeRpcDispatcherOptions) {
this._communicator = options.communicator;
@ -358,7 +449,7 @@ export class nopeDispatcher {
}
// Use the communicator to publish the event.
this._communicator.emitRpcResult(data.resultSink,result);
this._communicator.emitRpcResult(data.resultSink, result);
}
} catch (error) {
@ -448,15 +539,34 @@ export class nopeDispatcher {
// Subscribe to the availableServices of Remotes.
// If there is a new Service => udpate the External Services
this._communicator.onNewServicesAvailable((data: availableServices) => {
this._communicator.onNewServicesAvailable(data => {
try {
if (data.dispatcher !== this.id){
_this._mappingOfRemoteDispatchersAndFunctions.set(data.dispatcher, data);
_this._updateExternalServices();
}
if (data.dispatcher !== _this.id) {
_this._mappingOfRemoteDispatchersAndFunctions.set(data.dispatcher, data);
_this._updateExternalServices();
}
} catch (e) {
}
});
});
this._communicator.onNewTopicsAvailable(data => {
try {
if (data.dispatcher !== _this.id) {
_this._mappingOfRemoteDispatchersAndTopics.set(data.dispatcher, data);
_this._updateExternalTopics();
}
} catch (e) {
}
});
this._communicator.onBonjour((dispatcher: string) => {
if (_this.id !== dispatcher){
_this._sendAvailableServices();
_this._sendAvailableTopic();
}
});
this._communicator.emitBonjour(this.id);
}
/**
@ -484,7 +594,39 @@ export class nopeDispatcher {
4
)
);
}
}
}
/**
* Internal Function to update the Listing of external Topcis.
* This Function creates a list containing all subscriptions
* and publishers which are external.
*
* @protected
* @memberof nopeDispatcher
*/
protected _updateExternalTopics() {
const _this = this;
// Clear the Services
this._externalPublished.clear();
this._externalSubscribed.clear();
for (const dispatcherInfo of this._mappingOfRemoteDispatchersAndTopics.values()) {
dispatcherInfo.published.map(topic => _this._externalPublished.add(topic));
dispatcherInfo.subscribed.map(topic => _this._externalSubscribed.add(topic));
}
if (this._logger?.isDebugEnabled()) {
// If there is a Logger:
this._logger.debug(
'new topics found: \n' +
JSON.stringify(
Array.from(this._externalSubscribed),
undefined,
4
)
);
}
}
/**
@ -494,10 +636,14 @@ export class nopeDispatcher {
* @return {boolean} The result of the Test. True if either local or remotly a service is known.
* @memberof nopeDispatcher
*/
public serviceExists(id: string){
public serviceExists(id: string) {
return this._definedFunctions.has(id) || this._externalServices.has(id);
}
public subscriptionExists(topic: string){
return this._externalSubscribed.has(topic);
}
/**
* Function to adapt a Request name.
* Only used internally
@ -507,7 +653,7 @@ export class nopeDispatcher {
* @return {string} the adapted ID.
* @memberof nopeDispatcher
*/
protected _getName(id: string, type: 'request' | 'response'): string {
protected _getServiceName(id: string, type: 'request' | 'response'): string {
return id.startsWith(`${type}/`) ? id : `${type}/${id}`;
}
@ -521,7 +667,7 @@ export class nopeDispatcher {
* @memberof nopeDispatcher
*/
protected _subscribeToFunction(id: string, _cb: (...args) => Promise<any>): void {
const _req = this._getName(id, 'request');
const _req = this._getServiceName(id, 'request');
if (
this._subscriptionMode === 'individual' &&
!this._communicatorCallbacks.has(id)
@ -548,8 +694,9 @@ export class nopeDispatcher {
}
}
protected _subscribeToResult(id: string, deleteAfterCalling: boolean): void {
const _res = this._getName(id, 'response');
const _res = this._getServiceName(id, 'response');
if (
this._subscriptionMode === 'individual' &&
!this._communicatorCallbacks.has(id)
@ -561,7 +708,7 @@ export class nopeDispatcher {
const cb = (data: responseTaskMsg) => {
if (data.type === 'response') {
if (_this._handleExternalResponse(data)) {
_this._removeSubscription(id);
_this._removeRpcSubscription(id);
}
}
};
@ -650,7 +797,7 @@ export class nopeDispatcher {
} = {}): boolean {
const _id = typeof func === 'string' ? func : func['id'] as string || '0';
this._removeSubscription(_id);
this._removeRpcSubscription(_id);
if (!options.preventSendingToRegistery) {
// Publish the Available Services.
@ -660,7 +807,269 @@ export class nopeDispatcher {
return this._definedFunctions.delete(_id);
}
protected _removeSubscription(_id: string){
protected _externallySubscribeObservables: Map<string, {
observable: nopeObservable<externalEvent>,
cb: (...arg) => void,
}>;
protected _internallySubscribeObservables: Map<string, Set<nopeObservable<any>>>;
/**
* Creates an Event listener (if required)
*
* @protected
* @param {string} event The Event to Listen.
* @return {nopeObservable<externalEvent>} An Listener on the Communication Channel.
* @memberof nopeDispatcher
*/
protected _subscribeToEvent(event: string) {
const item = this._externallySubscribeObservables.get(event) || {
observable: new nopeObservable<externalEvent>(),
cb: () => { },
};
if (!item.observable.hasSubscriptions) {
const _this = this;
const cb = (data: externalEvent) => {
item.observable.setContent(data, _this.id);
};
this._communicator.onEvent(event, cb);
item.cb = cb;
}
// Set the Items.
this._externallySubscribeObservables.set(event, item);
return item.observable;
}
/**
* Function to unsubscribe from an event of the channel.
*
* @protected
* @param {string} event
* @memberof nopeDispatcher
*/
protected _unsubscribeEvent(event: string) {
const item = this._externallySubscribeObservables.get(event) || {
observable: new nopeObservable<any>(),
cb: () => { },
count: 0,
};
if (item && item.observable.hasSubscriptions) {
this._communicator.offEvent(event, item.cb);
this._externallySubscribeObservables.delete(event);
}
}
/**
* Function to register a Oberservable into the element.
*
* @template T
* @template K
* @template S
* @template G
* @param {nopeObservable<T,S,G>} observable
* @param {({
* mode: 'subscribe' | 'publish' | Array<'subscribe' | 'publish'>,
* topic: string | {
* subscribe?: string,
* publish?: string,
* },
* pipe?:{
* pipe?: pipe<externalEvent,K>,
* scope?: { [index: string]: any }
* },
* preventSendingToRegistery?: boolean
* })} options
* @return {*} {nopeObservable<T,S,G>}
* @memberof nopeDispatcher
*/
public registerObservable<T, K, S = T, G = T>(observable: nopeObservable<T, S, G>, options: {
mode: 'subscribe' | 'publish' | Array<'subscribe' | 'publish'>,
topic: string | {
subscribe?: string,
publish?: string,
},
pipe?: {
pipe?: pipe<externalEvent, K>,
scope?: { [index: string]: any }
},
preventSendingToRegistery?: boolean
}): nopeObservable<T, S, G> {
// Reference to itself
const _this = this;
// Extract the Topics, pipe and scope.
const _subTopic = typeof options.topic === 'string' ? options.topic : options.topic.subscribe || null;
const _pubTopic = typeof options.topic === 'string' ? options.topic : options.topic.publish || null;
const _pipe = typeof options.pipe === 'function' ? options.pipe || null : null;
const _scope = typeof options.pipe === 'object' ? options.pipe.scope || null : null
// A Flag, indicating, whether the topic is new or not.
let newElement = false;
// Test if the Item should be subscribe or not.
if (options.mode == 'subscribe' || (Array.isArray(options.mode) && options.mode.includes('subscribe'))) {
newElement = newElement || !this._externallySubscribeObservables.has(_subTopic);
const _externalSource = this._subscribeToEvent(_subTopic);
if (_pipe) {
const observer = _externalSource.enhancedSubscription((data: externalEvent) => {
// Test if the Content, which has been forwared in here inst the own dispathcer.
if (data.sender != _this.id){
observable.setContent(data.data, _this.id, data.timestamp);
}
}, {
scope: _scope,
pipe: _pipe
});
const dispose = observable.dispose;
observable.dispose = () => {
// Kill the Observer;
observer.unsubscribe();
// Unsubscribe the Event
_this._unsubscribeEvent(_subTopic);
// Call the original Dispose function;
dispose.apply(observable);
}
} else {
const observer = _externalSource.subscribe({
next(data: externalEvent) {
if (_this.id !== data.sender) {
observable.setContent(data.data, _this.id, data.timestamp);
}
},
complete() {
observable.observable.complete();
},
error(err) {
observable.observable.error(err);
}
});
// Overwrite the Original Dispose Function.
const dispose = observable.dispose;
observable.dispose = () => {
// Kill the Observer;
observer.unsubscribe();
// Unsubscribe the Event
_this._unsubscribeEvent(_subTopic);
// Call the original Dispose function;
dispose.apply(observable);
}
}
}
if (options.mode == 'publish' || (Array.isArray(options.mode) && options.mode.includes('publish'))) {
const cb = (data, sender?, timestamp?, ...args) => {
// Only Publish data, if there exists a Subscription.
if (_this.subscriptionExists(_pubTopic) && _this.id !== sender) {
// Use the Communicator to emit the Event.
_this._communicator.emitEvent(_pubTopic, {
data:data,
topic: _pubTopic,
sender: _this.id,
type: 'event',
timestamp
});
}
}
// Update the Flag.
newElement = newElement || !this._internallySubscribeObservables.has(_pubTopic);
// Register the Internally Subscribed Element.
const _set = this._internallySubscribeObservables.get(_pubTopic) || new Set();
_set.add(observable);
this._internallySubscribeObservables.set(_pubTopic, _set);
if (_pipe) {
const observer = observable.enhancedSubscription(cb, {
scope: _scope,
pipe: _pipe
});
// Overwrite the Original Dispose Function.
const dispose = observable.dispose;
observable.dispose = () => {
// Kill the Observer;
observer.unsubscribe();
// Unsubscribe the Event
_this._unsubscribeEvent(_subTopic);
// Unregister the Internally Subscribed Element.
const _set = _this._internallySubscribeObservables.get(_pubTopic) || new Set();
_set.delete(observable);
if (_set.size > 0) {
_this._internallySubscribeObservables.set(_pubTopic, _set);
} else {
_this._internallySubscribeObservables.delete(_pubTopic);
// Optionally send an update.
if (!options.preventSendingToRegistery) {
// Publish the Available Services.
_this._sendAvailableTopic();
}
}
// Call the original Dispose function;
dispose.apply(observable);
}
} else {
const observer = observable.subscribe(cb);
// Overwrite the Original Dispose Function.
const dispose = observable.dispose;
observable.dispose = () => {
// Kill the Observer;
observer.unsubscribe();
// Unsubscribe the Event
_this._unsubscribeEvent(_subTopic);
// Unregister the Internally Subscribed Element.
const _set = _this._internallySubscribeObservables.get(_pubTopic) || new Set();
_set.delete(observable);
if (_set.size > 0) {
_this._internallySubscribeObservables.set(_pubTopic, _set);
} else {
_this._internallySubscribeObservables.delete(_pubTopic);
// Optionally send an update.
if (!options.preventSendingToRegistery) {
// Publish the Available Services.
_this._sendAvailableTopic();
}
}
// Call the original Dispose function;
dispose.apply(observable);
}
}
}
if (!options.preventSendingToRegistery && newElement) {
// Publish the Available Services.
this._sendAvailableTopic();
}
// Return the Function.
return observable;
}
protected _removeRpcSubscription(_id: string) {
// Try to unregister the Callback from the communcator:
if (this._communicatorCallbacks.has(_id)) {
@ -701,6 +1110,25 @@ export class nopeDispatcher {
this._communicator.emitNewServicesAvailable(message);
}
/**
* Function to emit the available topics.
*
* @protected
* @memberof nopeDispatcher
*/
protected _sendAvailableTopic() {
// Define the Message
const message: availableTopics = {
dispatcher: this.id,
published: Array.from(this._internallySubscribeObservables.keys()),
subscribed: Array.from(this._externallySubscribeObservables.keys())
}
// Send the Message.
this._communicator.emitNewTopicsAvailable(message);
}
/**
* Function which is used to perform a call on the remote.
*
@ -727,7 +1155,7 @@ export class nopeDispatcher {
deletableCallbacks: [],
paramsHasNoCallback: false,
dynamicCallback: false,
resultSink: (this._resultSharing === 'generic' ? 'response' : this._getName(functionName, 'response')) as string
resultSink: (this._resultSharing === 'generic' ? 'response' : this._getServiceName(functionName, 'response')) as string
} as callOptions, options) as callOptions;
this._subscribeToResult(functionName, _options.dynamicCallback)
@ -792,7 +1220,7 @@ export class nopeDispatcher {
deleteAfterCalling,
dynamicCallback: true,
deletableCallbacks: [],
resultSink: _this._resultSharing === 'generic' ? 'response' : _this._getName(_func['id'], 'response')
resultSink: _this._resultSharing === 'generic' ? 'response' : _this._getServiceName(_func['id'], 'response')
});
}
}
@ -802,18 +1230,18 @@ export class nopeDispatcher {
// Create an Error:
const error = new Error('No Service Provider known for "' + functionName + '"');
if (_this._logger){
if (_this._logger) {
_this._logger.error('No Service Provider known for "' + functionName + '"');
_this._logger.error(error)
}
throw error;
throw error;
}
// Send the Message to the specific element:
if (_this._subscriptionMode === 'individual') {
_this._communicator.emitRpcRequest(_this._getName(taskRequest.functionId, 'request'), taskRequest);
_this._communicator.emitRpcRequest(_this._getServiceName(taskRequest.functionId, 'request'), taskRequest);
} else {
_this._communicator.emitRpcRequest('request', taskRequest);
}
@ -872,8 +1300,16 @@ export class nopeDispatcher {
*/
public reset(): void {
this._remotlyCalledFunctions = new Set();
this._externalServices = new Set();
this._mappingOfRemoteDispatchersAndFunctions = new Map();
this._mappingOfRemoteDispatchersAndTopics = new Map();
this._externalServices = new Set();
this._externalPublished = new Set();
this._externalSubscribed = new Set();
this._internallySubscribeObservables = new Map();
this._externallySubscribeObservables = new Map();
this.clearTasks();
this.unregisterAll();
}

28
test/testDispatcher.ts
View File

@ -1,9 +1,11 @@
import 'reflect-metadata';
import { EventLayer } from "../lib/communication/eventLayer";
import { getLinkedDispatcher } from "../lib/dispatcher/getLinkedDispatcher";
import { nopeDispatcher } from "../lib/dispatcher/nopeDispatcher";
import { exportFunctionToDispatcher } from "../lib/dispatcher/nopeDispatcherDecorators";
import { generateBenchmarkFunction } from "../modules/funcs/generateBenchmarkFunction";
import { nopeObservable } from '../lib/observables/nopeObservable';
let max = 100000;
const communicator = new EventLayer(
@ -20,12 +22,34 @@ const _benchmark = exportFunctionToDispatcher(generateBenchmarkFunction(max, '')
uri: 'benchmark'
});
const remote = getLinkedDispatcher({ communicator });
const remoteObservable = new nopeObservable<number>()
remote.registerObservable(remoteObservable,{
mode: 'publish',
topic: 'topic'
});
const main = async () => {
const localObservable = new nopeObservable<number>();
localObservable.subscribe((data) => {
console.log(data)
});
remoteObservable.setContent(1337);
console.log('No output of the emitter');
local.registerObservable(localObservable,{
mode: 'subscribe',
topic:'topic'
});
remoteObservable.setContent(1338)
try {
await local.performCall('unkown', []);
} catch (e) {
console.log('Succes');
console.log('Successfully thrown Error. Function isnt available.');
console.error(e)
}