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

Adding onPublishersChanged, and publishers

Browse Source
This commit is contained in:
Martin Karkowski 2021-11-17 07:27:02 +01:00
parent bca4898e28
commit 5abe77f92c
2 changed files with 91 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

104
lib/pubSub/nopePubSubSystem.ts
View File

@ -49,6 +49,13 @@ export class PubSubSystemBase<I extends INopeEventEmitter = INopeEventEmitter, O
readonly subscriptions: INopeObservable<string[]>;
readonly onPublishersChanged: INopeEventEmitter<{
added: string[],
removed: string[]
}>;
readonly publishers: INopeObservable<string[]>;
protected _comparePatternAndPath: TcomparePatternAndPathFunc;
/**
@ -106,27 +113,72 @@ export class PubSubSystemBase<I extends INopeEventEmitter = INopeEventEmitter, O
this.subscriptions = new NopeObservable();
this.subscriptions.setContent([]);
this.publishers = new NopeObservable();
this.publishers.setContent([]);
this.onIncrementalDataChange = new NopeEventEmitter();
this.onSubscriptionChanged = new NopeEventEmitter();
this.onPublishersChanged = new NopeEventEmitter();
}
protected _determineSusbcriptions(): Set<string> {
protected _updateEmitters(beforeAdding: {
publishers: Set<string>,
subscriptions: Set<string>
}): void {
const afterAdding = this._determineCurrentPublishersAndSubscribers();
const diffSubscribers = determineDifference(beforeAdding.subscriptions, afterAdding.subscriptions);
const diffPublishers = determineDifference(beforeAdding.subscriptions, afterAdding.subscriptions);
if (diffSubscribers.removed.size > 0 || diffSubscribers.added.size > 0) {
// Update the currently used subscriptions
this.subscriptions.setContent(
Array.from(afterAdding.subscriptions)
);
// Now emit, that there is a new subscription.
this.onSubscriptionChanged.emit({
added: Array.from(diffSubscribers.added),
removed: Array.from(diffSubscribers.removed)
});
}
if (diffPublishers.removed.size > 0 || diffPublishers.added.size > 0) {
// Update the currently used subscriptions
this.publishers.setContent(
Array.from(afterAdding.publishers)
);
// Now emit, that there is a new subscription.
this.onPublishersChanged.emit({
added: Array.from(diffPublishers.added),
removed: Array.from(diffPublishers.removed)
});
}
}
protected _determineCurrentPublishersAndSubscribers(): {
publishers: Set<string>,
subscriptions: Set<string>
} {
const subscriptions = new Set<string>();
const publishers = new Set<string>();
for (const { subTopic } of this._emitters.values()) {
for (const { subTopic, pubTopic } of this._emitters.values()) {
if (subTopic) {
subscriptions.add(subTopic);
}
if (pubTopic) {
publishers.add(pubTopic);
}
}
return subscriptions;
return {
publishers,
subscriptions
};
}
public register(emitter: I, options: IPropertyOptions): O {
if (!this._emitters.has(emitter as unknown as O)) {
const subscriptionsBeforeAdding = this._determineSusbcriptions();
const beforeAdding = this._determineCurrentPublishersAndSubscribers();
let pubTopic: string | false =
typeof options.topic === "string"
@ -175,21 +227,10 @@ export class PubSubSystemBase<I extends INopeEventEmitter = INopeEventEmitter, O
});
}
this._updateEmitters(beforeAdding);
// Now, if required, add the Data to the emitter.
if (subTopic) {
const subscriptionsAfterAdding = this._determineSusbcriptions();
const diff = determineDifference(subscriptionsBeforeAdding, subscriptionsAfterAdding);
if (diff.removed.size > 0 || diff.added.size > 0) {
this.subscriptions.setContent(Array.from(subscriptionsAfterAdding));
this.onSubscriptionChanged.emit({
added: Array.from(diff.added),
removed: Array.from(diff.removed)
});
}
if (this._sendCurrentDataOnSubscription) {
if (subTopic && this._sendCurrentDataOnSubscription) {
if (containsWildcards(subTopic)) {
// This is more Complex.
this._patternbasedPullData(subTopic, null).map(item => {
@ -220,7 +261,6 @@ export class PubSubSystemBase<I extends INopeEventEmitter = INopeEventEmitter, O
}
}
}
} else {
throw Error("Already registered Emitter!");
@ -231,7 +271,7 @@ export class PubSubSystemBase<I extends INopeEventEmitter = INopeEventEmitter, O
public updateOptions(emitter: I, options: IPropertyOptions): void {
if (this._emitters.has(emitter as unknown as O)) {
const subscriptionsBeforeAdding = this._determineSusbcriptions();
const beforeAdding = this._determineCurrentPublishersAndSubscribers();
const pubTopic =
typeof options.topic === "string"
@ -249,29 +289,13 @@ export class PubSubSystemBase<I extends INopeEventEmitter = INopeEventEmitter, O
data.subTopic = subTopic;
data.pubTopic = pubTopic;
this._emitters.set(emitter as unknown as O, data);
if (subTopic) {
const subscriptionsAfterAdding = this._determineSusbcriptions();
const diff = determineDifference(subscriptionsBeforeAdding, subscriptionsAfterAdding);
if (diff.removed.size > 0 || diff.added.size > 0) {
// Update the currently used subscriptions
this.subscriptions.setContent(
Array.from(subscriptionsAfterAdding)
);
// Now emit, that there is a new subscription.
this.onSubscriptionChanged.emit({
added: Array.from(diff.added),
removed: Array.from(diff.removed)
});
}
}
// Update the Matching Rules.
this.updateMatchting();
this._updateEmitters(beforeAdding);
} else {
throw Error("Already registered Emitter!");
}

7
lib/types/nope/nopePubSub.interface.ts
View File

@ -159,6 +159,13 @@ export interface IPubSubSystem<I extends INopeEventEmitter = INopeEventEmitter,
added: string[],
removed: string[],
}>
readonly publishers: INopeObservable<string[]>;
readonly onPublishersChanged: INopeEventEmitter<{
added: string[],
removed: string[],
}>
}
/**