|
|
@@ -3,10 +3,36 @@ import {sha256} from "./utils";
|
|
|
|
|
|
class ConnectionManager {
|
|
|
constructor() {
|
|
|
+ console.log(`connection manager`)
|
|
|
this.connections = new Map(); // ConnectionId -> Connection (hash, see below)
|
|
|
+ this.stateListener = null; // Fired when connection state changes
|
|
|
+ this.notificationListener = null; // Fired when new notifications arrive
|
|
|
}
|
|
|
|
|
|
- async refresh(subscriptions, users, onNotification) {
|
|
|
+ registerStateListener(listener) {
|
|
|
+ this.stateListener = listener;
|
|
|
+ }
|
|
|
+
|
|
|
+ resetStateListener() {
|
|
|
+ this.stateListener = null;
|
|
|
+ }
|
|
|
+
|
|
|
+ registerNotificationListener(listener) {
|
|
|
+ this.notificationListener = listener;
|
|
|
+ }
|
|
|
+
|
|
|
+ resetNotificationListener() {
|
|
|
+ this.notificationListener = null;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * This function figures out which websocket connections should be running by comparing the
|
|
|
+ * current state of the world (connections) with the target state (targetIds).
|
|
|
+ *
|
|
|
+ * It uses a "connectionId", which is sha256($subscriptionId|$username|$password) to identify
|
|
|
+ * connections. If any of them change, the connection is closed/replaced.
|
|
|
+ */
|
|
|
+ async refresh(subscriptions, users) {
|
|
|
if (!subscriptions || !users) {
|
|
|
return;
|
|
|
}
|
|
|
@@ -17,10 +43,9 @@ class ConnectionManager {
|
|
|
const connectionId = await makeConnectionId(s, user);
|
|
|
return {...s, user, connectionId};
|
|
|
}));
|
|
|
- const activeIds = subscriptionsWithUsersAndConnectionId.map(s => s.connectionId);
|
|
|
- const deletedIds = Array.from(this.connections.keys()).filter(id => !activeIds.includes(id));
|
|
|
+ const targetIds = subscriptionsWithUsersAndConnectionId.map(s => s.connectionId);
|
|
|
+ const deletedIds = Array.from(this.connections.keys()).filter(id => !targetIds.includes(id));
|
|
|
|
|
|
- console.log(subscriptionsWithUsersAndConnectionId);
|
|
|
// Create and add new connections
|
|
|
subscriptionsWithUsersAndConnectionId.forEach(subscription => {
|
|
|
const subscriptionId = subscription.id;
|
|
|
@@ -31,7 +56,16 @@ class ConnectionManager {
|
|
|
const topic = subscription.topic;
|
|
|
const user = subscription.user;
|
|
|
const since = subscription.last;
|
|
|
- const connection = new Connection(connectionId, subscriptionId, baseUrl, topic, user, since, onNotification);
|
|
|
+ const connection = new Connection(
|
|
|
+ connectionId,
|
|
|
+ subscriptionId,
|
|
|
+ baseUrl,
|
|
|
+ topic,
|
|
|
+ user,
|
|
|
+ since,
|
|
|
+ (subscriptionId, notification) => this.notificationReceived(subscriptionId, notification),
|
|
|
+ (subscriptionId, state) => this.stateChanged(subscriptionId, state)
|
|
|
+ );
|
|
|
this.connections.set(connectionId, connection);
|
|
|
console.log(`[ConnectionManager] Starting new connection ${connectionId} (subscription ${subscriptionId} with user ${user ? user.username : "anonymous"})`);
|
|
|
connection.start();
|
|
|
@@ -46,6 +80,18 @@ class ConnectionManager {
|
|
|
connection.close();
|
|
|
});
|
|
|
}
|
|
|
+
|
|
|
+ stateChanged(subscriptionId, state) {
|
|
|
+ if (this.stateListener) {
|
|
|
+ this.stateListener(subscriptionId, state);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ notificationReceived(subscriptionId, notification) {
|
|
|
+ if (this.notificationListener) {
|
|
|
+ this.notificationListener(subscriptionId, notification);
|
|
|
+ }
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
const makeConnectionId = async (subscription, user) => {
|