Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Generates a unique subscription key per subscriber #3

Open
wants to merge 3 commits into
base: kai/topic-config-file-param
Choose a base branch
from
Open

Generates a unique subscription key per subscriber #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2430c0c
Generates a unique subscription key per subscriber
kaikoehn Sep 11, 2020
f2d2090
Merge branch 'develop' into kai/multi-subscriber-fix
meyerj Feb 20, 2022
8c3bce9
Merge branch 'kai/topic-config-file-param' into kai/multi-subscriber-fix
meyerj Jul 20, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions lib/resource_provider.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ class ResourceProvider {
}

getSubscriptionByTopicName(topicName) {
return SubscriptionManager.getInstance().getSubscriptionByTopicName(topicName).get();
return SubscriptionManager.getInstance().getSubscriptionByTopicName(topicName, this._bridgeId).get();
}

getClientByServiceName(serviceName) {
Expand Down Expand Up @@ -126,7 +126,7 @@ class ResourceProvider {
}

hasSubscription(topicName) {
return SubscriptionManager.getInstance().getSubscriptionByTopicName(topicName) !== undefined;
return SubscriptionManager.getInstance().getSubscriptionByTopicName(topicName, this._bridgeId) !== undefined;
}

clean() {
Expand Down
28 changes: 18 additions & 10 deletions lib/subscription_manager.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,12 +46,18 @@ class SubscriptionManager {
this._node = node;
}

getSubscriptionByTopicName(topicName) {
return this._subscripions.get(topicName);
getSubscriptionByTopicName(topicName, bridgeId) {
const subscriptionKey = this.getSubscriptionKey(topicName, bridgeId)
return this._subscripions.get(subscriptionKey);
}

getSubscriptionKey(topicName, bridgeId) {
return `${topicName}:${bridgeId}`;
}

createSubscription(messageType, topicName, bridgeId, callback) {
let handle = this._subscripions.get(topicName);
const subscriptionKey = this.getSubscriptionKey(topicName, bridgeId)
let handle = this._subscripions.get(subscriptionKey);

if (!handle) {
const defaultOpts = {}
Expand All @@ -60,13 +66,13 @@ class SubscriptionManager {
console.log(topicName, opts)

let subscription = this._node.createSubscription(messageType, topicName, {enableTypedArray: false, ...opts}, (message) => {
this._subscripions.get(topicName).callbacks.forEach(callback => {
this._subscripions.get(subscriptionKey).callbacks.forEach(callback => {
callback(topicName, message);
});
});
handle = new HandleWithCallbacks(subscription, this._node.destroySubscription.bind(this._node));
handle.addCallback(bridgeId, callback);
this._subscripions.set(topicName, handle);
this._subscripions.set(subscriptionKey, handle);
debug(`Subscription has been created, and the topic name is ${topicName}.`);

return handle.get();
Expand All @@ -78,13 +84,15 @@ class SubscriptionManager {
}

destroySubscription(topicName, bridgeId) {
if (this._subscripions.has(topicName)) {
let handle = this._subscripions.get(topicName);
const subscriptionKey = this.getSubscriptionKey(topicName, bridgeId)

if (this._subscripions.has(subscriptionKey)) {
let handle = this._subscripions.get(subscriptionKey);
if (handle.hasCallbackForId(bridgeId)) {
handle.removeCallback(bridgeId);
handle.release();
if (handle.count === 0) {
this._subscripions.delete(topicName);
this._subscripions.delete(subscriptionKey);
}
}
}
Expand All @@ -101,9 +109,9 @@ class SubscriptionManager {
}

_removeInvalidHandle() {
this._subscripions.forEach((handle, topicName, map) => {
this._subscripions.forEach((handle, subscriptionKey, map) => {
if (handle.count === 0) {
map.delete(topicName);
map.delete(subscriptionKey);
}
});
}
Expand Down