chore: make topic mandatory for signals

Signed-off-by: Heikki Hellgren <heikki.hellgren@op.fi>

packages/backend/src/plugins/signals.ts

@@ -29,7 +29,7 @@ export default async function createPlugin(
 
   setInterval(() => {
     console.log('publishing');
-    service.publish('*', { hello: 'world' });
+    service.publish('*', 'devtools:info', { now: new Date().toISOString() });
   }, 5000);
 
   return await createRouter({

plugins/devtools/src/components/Content/InfoContent/InfoContent.tsx

@@ -77,7 +77,7 @@ export const InfoContent = () => {
   const { about, loading, error } = useInfo();
   // Just testing for signals
   const [messages, setMessages] = React.useState<string[]>([]);
-  useSignalsApi(message => {
+  useSignalsApi('devtools:info', message => {
     messages.push(JSON.stringify(message));
     setMessages([...messages]);
   });

plugins/signals-node/api-report.md

@@ -29,15 +29,13 @@ export type SignalsEventBrokerPayload = {
 export class SignalsService implements EventSubscriber {
   // (undocumented)
   static create(options: ServiceOptions): SignalsService;
-  // (undocumented)
   handleUpgrade: (req: Request_2) => Promise<void>;
   // (undocumented)
   onEvent(params: EventParams<SignalsEventBrokerPayload>): Promise<void>;
-  // (undocumented)
   publish(
     to: string | string[],
+    topic: string,
     message: JsonObject,
-    topic?: string,
   ): Promise<void>;
   // (undocumented)
   supportsEventTopics(): string[];

plugins/signals-node/src/SignalsService.ts

@@ -76,6 +76,11 @@ export class SignalsService implements EventSubscriber {
     this.eventBroker?.subscribe(this);
   }
 
+  /**
+   * Handles request upgradce to websocket and adds the connection to internal
+   * list for publish/subscribe functionality
+   * @param req - Request
+   */
   handleUpgrade = async (req: Request) => {
     const identity = await this.identity.getIdentity({
       request: req,
@@ -157,23 +162,29 @@ export class SignalsService implements EventSubscriber {
     }
   }
 
-  async publish(to: string | string[], message: JsonObject, topic?: string) {
+  /**
+   * Publishes a message to user refs to specific topic
+   * @param to - string or array of user ref strings to publish message to
+   * @param topic - message topic
+   * @param message - message to publish
+   */
+  async publish(to: string | string[], topic: string, message: JsonObject) {
     await this.publishInternal(
       Array.isArray(to) ? to : [to],
+      topic,
       message,
       false,
-      topic,
     );
   }
 
   private async publishInternal(
     recipients: string[],
+    topic: string,
     message: JsonObject,
     brokedEvent: boolean,
-    topic?: string,
   ) {
     this.connections.forEach(conn => {
-      if (topic && !conn.subscriptions.has(topic)) {
+      if (!conn.subscriptions.has(topic)) {
         return;
       }
       // Sending to all users can be done with '*'
@@ -208,15 +219,19 @@ export class SignalsService implements EventSubscriber {
       return;
     }
 
-    if (!eventPayload?.recipients || !eventPayload.message) {
+    if (
+      !eventPayload?.recipients ||
+      !eventPayload.topic ||
+      !eventPayload.message
+    ) {
       return;
     }
 
     await this.publishInternal(
       eventPayload.recipients,
+      eventPayload.topic,
       eventPayload.message,
       true,
-      eventPayload.topic,
     );
   }
 

plugins/signals-react/api-report.md

@@ -12,9 +12,9 @@ import { JsonObject } from '@backstage/types';
 export type SignalsApi = {
   subscribe(
     onMessage: (message: JsonObject, topic?: string) => void,
-    topic?: string,
+    topic: string,
   ): void;
-  unsubscribe(topic?: string): void;
+  unsubscribe(topic: string): void;
 };
 
 // @public (undocumented)
@@ -27,18 +27,15 @@ export class SignalsClient implements SignalsApi {
   // (undocumented)
   static instance: SignalsClient | null;
   // (undocumented)
-  subscribe(
-    onMessage: (message: JsonObject, topic?: string) => void,
-    topic?: string,
-  ): void;
+  subscribe(onMessage: (message: JsonObject) => void, topic: string): void;
   // (undocumented)
-  unsubscribe(topic?: string): void;
+  unsubscribe(topic: string): void;
 }
 
 // @public (undocumented)
 export const useSignalsApi: (
+  topic: string,
   onMessage: (message: JSONObject) => void,
-  topic?: string,
 ) => void;
 
 // (No @packageDocumentation comment for this package)

plugins/signals-react/src/api/SignalsApi.ts

@@ -25,8 +25,8 @@ export const signalsApiRef = createApiRef<SignalsApi>({
 export type SignalsApi = {
   subscribe(
     onMessage: (message: JsonObject, topic?: string) => void,
-    topic?: string,
+    topic: string,
   ): void;
 
-  unsubscribe(topic?: string): void;
+  unsubscribe(topic: string): void;
 };

plugins/signals-react/src/api/SignalsClient.ts

@@ -22,8 +22,7 @@ export class SignalsClient implements SignalsApi {
   static instance: SignalsClient | null = null;
   private ws: WebSocket | null = null;
   private discoveryApi: DiscoveryApi;
-  private cbs: Map<string, (message: JsonObject, topic?: string) => void> =
-    new Map();
+  private cbs: Map<string, (message: JsonObject) => void> = new Map();
   private queue: JsonObject[] = [];
   private reconnectTimeout: any;
 
@@ -38,25 +37,20 @@ export class SignalsClient implements SignalsApi {
     this.discoveryApi = options.discoveryApi;
   }
 
-  subscribe(
-    onMessage: (message: JsonObject, topic?: string) => void,
-    topic?: string,
-  ): void {
-    const subscriptionTopic = topic ?? '*';
+  subscribe(onMessage: (message: JsonObject) => void, topic: string): void {
     // Do not allow to subscribe to same topic multiple times
-    if (this.cbs.has(subscriptionTopic)) {
+    if (this.cbs.has(topic)) {
       return;
     }
 
-    this.cbs.set(subscriptionTopic, onMessage);
+    this.cbs.set(topic, onMessage);
     this.connect().then(() => {
       this.send({ action: 'subscribe', topic });
     });
   }
 
-  unsubscribe(topic?: string): void {
-    const subscriptionTopic = topic ?? '*';
-    this.cbs.delete(subscriptionTopic);
+  unsubscribe(topic: string): void {
+    this.cbs.delete(topic);
     this.send({ action: 'unsubscribe', topic });
   }
 
@@ -91,12 +85,11 @@ export class SignalsClient implements SignalsApi {
     this.ws.onmessage = (data: MessageEvent) => {
       try {
         const json = JSON.parse(data.data) as JsonObject;
-        let cb = this.cbs.get('*');
         if (json.topic) {
-          cb = this.cbs.get(json.topic as string);
-        }
-        if (cb) {
-          cb(json.message as JsonObject, json.topic as string);
+          const cb = this.cbs.get(json.topic as string);
+          if (cb) {
+            cb(json.message as JsonObject);
+          }
         }
       } catch (e) {
         // NOOP
@@ -127,7 +120,12 @@ export class SignalsClient implements SignalsApi {
         this.ws.close();
       }
       this.ws = null;
-      this.connect();
+      this.connect().then(() => {
+        // Resubscribe to existing topics in case we lost connection
+        for (const topic of this.cbs.keys()) {
+          this.send({ action: 'subscribe', topic });
+        }
+      });
     }, 5000);
   }
 }

plugins/signals-react/src/hooks/useSignalsApi.ts

@@ -20,8 +20,8 @@ import { useEffect } from 'react';
 
 /** @public */
 export const useSignalsApi = (
+  topic: string,
   onMessage: (message: JSONObject) => void,
-  topic?: string,
 ) => {
   const discovery = useApi(discoveryApiRef);
   const signals = SignalsClient.create({ discoveryApi: discovery });