diff --git a/packages/backend/src/index.ts b/packages/backend/src/index.ts index 0e92698230..bade028597 100644 --- a/packages/backend/src/index.ts +++ b/packages/backend/src/index.ts @@ -101,9 +101,7 @@ function makeCreateEnv(config: Config) { const eventBroker = new DefaultEventBroker(root.child({ type: 'plugin' })); const signalService = DefaultSignalService.create({ - logger: root, eventBroker, - identity, }); root.info(`Created UrlReader ${reader}`); diff --git a/packages/backend/src/plugins/signals.ts b/packages/backend/src/plugins/signals.ts index 687fcfaa33..477cea1938 100644 --- a/packages/backend/src/plugins/signals.ts +++ b/packages/backend/src/plugins/signals.ts @@ -22,6 +22,7 @@ export default async function createPlugin( ): Promise { return await createRouter({ logger: env.logger, - service: env.signalService, + eventBroker: env.eventBroker, + identity: env.identity, }); } diff --git a/plugins/signals-backend/README.md b/plugins/signals-backend/README.md index f9ebfb80c7..e30a0836e0 100644 --- a/plugins/signals-backend/README.md +++ b/plugins/signals-backend/README.md @@ -20,7 +20,8 @@ export default async function createPlugin( ): Promise { return await createRouter({ logger: env.logger, - service: env.signalsService, + eventBroker: env.eventBroker, + identity: env.identity, }); } ``` @@ -29,7 +30,7 @@ Now add the signals to `packages/backend/src/index.ts`: ```ts // ... -import signals from './plugins/sonarqube'; +import signals from './plugins/signals'; async function main() { // ... diff --git a/plugins/signals-backend/api-report.md b/plugins/signals-backend/api-report.md index e96dd2e09c..dd46778e5d 100644 --- a/plugins/signals-backend/api-report.md +++ b/plugins/signals-backend/api-report.md @@ -4,9 +4,10 @@ ```ts import { BackendFeature } from '@backstage/backend-plugin-api'; +import { EventBroker } from '@backstage/plugin-events-node'; import express from 'express'; +import { IdentityApi } from '@backstage/plugin-auth-node'; import { LoggerService } from '@backstage/backend-plugin-api'; -import { SignalService } from '@backstage/plugin-signals-node'; // @public (undocumented) export function createRouter(options: RouterOptions): Promise; @@ -14,9 +15,11 @@ export function createRouter(options: RouterOptions): Promise; // @public (undocumented) export interface RouterOptions { // (undocumented) - logger: LoggerService; + eventBroker?: EventBroker; // (undocumented) - service: SignalService; + identity: IdentityApi; + // (undocumented) + logger: LoggerService; } // @public diff --git a/plugins/signals-backend/src/plugin.ts b/plugins/signals-backend/src/plugin.ts index 68bbf27377..4a9cef028f 100644 --- a/plugins/signals-backend/src/plugin.ts +++ b/plugins/signals-backend/src/plugin.ts @@ -18,7 +18,6 @@ import { createBackendPlugin, } from '@backstage/backend-plugin-api'; import { createRouter } from './service/router'; -import { signalService } from '@backstage/plugin-signals-node'; /** * Signals backend plugin @@ -32,13 +31,15 @@ export const signalsPlugin = createBackendPlugin({ deps: { httpRouter: coreServices.httpRouter, logger: coreServices.logger, - service: signalService, + identity: coreServices.identity, + // TODO: EventBroker. It is optional for now but it's actually required so waiting for the new backend system + // for the events-backend for this to work. }, - async init({ httpRouter, logger, service }) { + async init({ httpRouter, logger, identity }) { httpRouter.use( await createRouter({ logger, - service, + identity, }), ); }, diff --git a/plugins/signals-backend/src/service/SignalManager.ts b/plugins/signals-backend/src/service/SignalManager.ts new file mode 100644 index 0000000000..988db47465 --- /dev/null +++ b/plugins/signals-backend/src/service/SignalManager.ts @@ -0,0 +1,213 @@ +/* + * Copyright 2023 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +import { EventBroker, EventParams } from '@backstage/plugin-events-node'; +import { SignalPayload } from '@backstage/plugin-signals-node'; +import { RawData, WebSocket, WebSocketServer } from 'ws'; +import { IncomingMessage } from 'http'; +import { v4 as uuid } from 'uuid'; +import { JsonObject } from '@backstage/types'; +import { + BackstageIdentityResponse, + IdentityApi, + IdentityApiGetIdentityRequest, +} from '@backstage/plugin-auth-node'; +import { LoggerService } from '@backstage/backend-plugin-api'; +import { Duplex } from 'stream'; + +/** @internal */ +export type ConnectionUpgradeOptions = { + request: IncomingMessage; + socket: Duplex; + head: Buffer; +}; + +/** + * @internal + */ +export type SignalConnection = { + id: string; + user: string; + ws: WebSocket; + ownershipEntityRefs: string[]; + subscriptions: Set; +}; + +/** + * @internal + */ +export type SignalManagerOptions = { + // TODO: Remove optional when events-backend can offer this service + eventBroker?: EventBroker; + logger: LoggerService; + identity: IdentityApi; +}; + +/** @internal */ +export class SignalManager { + private connections: Map = new Map< + string, + SignalConnection + >(); + private eventBroker?: EventBroker; + private logger: LoggerService; + private identity: IdentityApi; + private server: WebSocketServer; + + static create(options: SignalManagerOptions) { + return new SignalManager(options); + } + + private constructor(options: SignalManagerOptions) { + ({ + eventBroker: this.eventBroker, + logger: this.logger, + identity: this.identity, + } = options); + + this.server = new WebSocketServer({ + noServer: true, + clientTracking: false, + }); + + this.eventBroker?.subscribe({ + supportsEventTopics: () => ['signals'], + onEvent: (params: EventParams) => + this.onEventBrokerEvent(params), + }); + } + + /** + * Handles request upgrade to websocket and adds the connection to internal + * list for publish/subscribe functionality + * @param req - Request + */ + async handleUpgrade(options: ConnectionUpgradeOptions) { + const { request, socket, head } = options; + let identity: BackstageIdentityResponse | undefined = undefined; + + // Authentication token is passed in Sec-WebSocket-Protocol header as there + // is no other way to pass the token with plain websockets + const token = request.headers['sec-websocket-protocol']; + if (token) { + identity = await this.identity.getIdentity({ + request: { + headers: { authorization: token }, + }, + } as IdentityApiGetIdentityRequest); + } + + this.server.handleUpgrade( + request, + socket, + head, + (ws: WebSocket, __: IncomingMessage) => { + this.addConnection(ws, identity); + }, + ); + } + + private addConnection(ws: WebSocket, identity?: BackstageIdentityResponse) { + const id = uuid(); + + const conn = { + id, + user: identity?.identity.userEntityRef ?? 'user:default/guest', + ws, + ownershipEntityRefs: identity?.identity.ownershipEntityRefs ?? [], + subscriptions: new Set(), + }; + + this.connections.set(id, conn); + + ws.on('error', (err: Error) => { + this.logger.info( + `Error occurred with connection ${id}: ${err}, closing connection`, + ); + ws.close(); + this.connections.delete(id); + }); + + ws.on('close', (code: number, reason: Buffer) => { + this.logger.info( + `Connection ${id} closed with code ${code}, reason: ${reason}`, + ); + this.connections.delete(id); + }); + + ws.on('message', (data: RawData, isBinary: boolean) => { + this.logger.debug(`Received message from connection ${id}: ${data}`); + if (isBinary) { + return; + } + try { + const json = JSON.parse(data.toString()) as JsonObject; + this.handleMessage(conn, json); + } catch (err: any) { + this.logger.error( + `Invalid message received from connection ${id}: ${err}`, + ); + } + }); + } + + private handleMessage(connection: SignalConnection, message: JsonObject) { + if (message.action === 'subscribe' && message.channel) { + this.logger.info( + `Connection ${connection.id} subscribed to ${message.channel}`, + ); + connection.subscriptions.add(message.channel as string); + } else if (message.action === 'unsubscribe' && message.channel) { + this.logger.info( + `Connection ${connection.id} unsubscribed from ${message.channel}`, + ); + connection.subscriptions.delete(message.channel as string); + } + } + + private async onEventBrokerEvent( + params: EventParams, + ): Promise { + const { eventPayload } = params; + if (!eventPayload.channel || !eventPayload.message) { + return; + } + + const { channel, recipients, message } = eventPayload; + const jsonMessage = JSON.stringify({ channel, message }); + + // Actual websocket message sending + this.connections.forEach(conn => { + if (!conn.subscriptions.has(channel)) { + return; + } + // Sending to all users can be done with null + if ( + recipients !== null && + !conn.ownershipEntityRefs.some((ref: string) => + recipients.includes(ref), + ) + ) { + return; + } + + if (conn.ws.readyState !== WebSocket.OPEN) { + return; + } + + conn.ws.send(jsonMessage); + }); + } +} diff --git a/plugins/signals-backend/src/service/router.test.ts b/plugins/signals-backend/src/service/router.test.ts index 281d6ed4a5..ecf2ad4aa6 100644 --- a/plugins/signals-backend/src/service/router.test.ts +++ b/plugins/signals-backend/src/service/router.test.ts @@ -18,9 +18,17 @@ import express from 'express'; import request from 'supertest'; import { createRouter } from './router'; -import { DefaultSignalService } from '@backstage/plugin-signals-node'; +import { EventBroker } from '@backstage/plugin-events-node'; +import { IdentityApi } from '@backstage/plugin-auth-node'; -const signalsServiceMock: jest.Mocked = {} as any; +const eventBrokerMock: jest.Mocked = { + subscribe: jest.fn(), + publish: jest.fn(), +}; + +const identityApiMock: jest.Mocked = { + getIdentity: jest.fn(), +}; describe('createRouter', () => { let app: express.Express; @@ -28,7 +36,8 @@ describe('createRouter', () => { beforeAll(async () => { const router = await createRouter({ logger: getVoidLogger(), - service: signalsServiceMock, + identity: identityApiMock, + eventBroker: eventBrokerMock, }); app = express().use(router); }); diff --git a/plugins/signals-backend/src/service/router.ts b/plugins/signals-backend/src/service/router.ts index 28bafce14a..ec56860bf3 100644 --- a/plugins/signals-backend/src/service/router.ts +++ b/plugins/signals-backend/src/service/router.ts @@ -17,22 +17,27 @@ import { errorHandler } from '@backstage/backend-common'; import express, { NextFunction, Request, Response } from 'express'; import Router from 'express-promise-router'; import { LoggerService } from '@backstage/backend-plugin-api'; -import { SignalService } from '@backstage/plugin-signals-node'; import * as https from 'https'; import http from 'http'; +import { SignalManager } from './SignalManager'; +import { IdentityApi } from '@backstage/plugin-auth-node'; +import { EventBroker } from '@backstage/plugin-events-node'; /** @public */ export interface RouterOptions { logger: LoggerService; - service: SignalService; + eventBroker?: EventBroker; + identity: IdentityApi; } /** @public */ export async function createRouter( options: RouterOptions, ): Promise { - const { logger, service } = options; + const { logger } = options; + const manager = SignalManager.create(options); let subscribed = false; + const upgradeMiddleware = (req: Request, _: Response, next: NextFunction) => { const server: https.Server | http.Server = (req.socket as any)?.server; if ( @@ -50,7 +55,7 @@ export async function createRouter( server.on('upgrade', async (request, socket, head) => { // TODO: Find a way to make this more generic if (request.url === '/api/signals') { - await service.handleUpgrade({ server, request, socket, head }); + await manager.handleUpgrade({ request, socket, head }); } }); } diff --git a/plugins/signals-backend/src/service/standaloneServer.ts b/plugins/signals-backend/src/service/standaloneServer.ts index e1ce27aee4..278de8621e 100644 --- a/plugins/signals-backend/src/service/standaloneServer.ts +++ b/plugins/signals-backend/src/service/standaloneServer.ts @@ -23,6 +23,11 @@ import { Logger } from 'winston'; import { createRouter } from './router'; import { DefaultSignalService } from '@backstage/plugin-signals-node'; import { DefaultIdentityClient } from '@backstage/plugin-auth-node'; +import { + EventBroker, + EventParams, + EventSubscriber, +} from '@backstage/plugin-events-node'; export interface ServerOptions { port: number; @@ -43,14 +48,26 @@ export async function startStandaloneServer( issuer: await discovery.getExternalBaseUrl('auth'), }); + const mockSubscribers: EventSubscriber[] = []; + const eventBroker: EventBroker = { + async publish(params: EventParams): Promise { + mockSubscribers.forEach(sub => sub.onEvent(params)); + }, + subscribe(...subscribers: EventSubscriber[]) { + subscribers.flat().forEach(subscriber => { + mockSubscribers.push(subscriber); + }); + }, + }; + const signals = DefaultSignalService.create({ - logger: logger, - identity, + eventBroker, }); const router = await createRouter({ logger, - service: signals, + identity, + eventBroker, }); let service = createServiceBuilder(module) @@ -60,10 +77,22 @@ export async function startStandaloneServer( service = service.enableCors({ origin: 'http://localhost:3000' }); } - return await service.start().catch(err => { + let server: Promise; + try { + server = service.start(); + + setInterval(() => { + signals.publish({ + recipients: null, + channel: 'test', + message: { hello: 'world' }, + }); + }, 5000); + } catch (err) { logger.error(err); process.exit(1); - }); + } + return server; } module.hot?.accept(); diff --git a/plugins/signals-node/README.md b/plugins/signals-node/README.md index 14c9332dc1..5e738a849d 100644 --- a/plugins/signals-node/README.md +++ b/plugins/signals-node/README.md @@ -27,10 +27,8 @@ function makeCreateEnv(config: Config) { // ... const eventBroker = new DefaultEventBroker(root.child({ type: 'plugin' })); - const signalService = SignalService.create({ - logger: root, - eventBroker, // EventBroker is optional - identity, + const signalService = DefaultSignalService.create({ + eventBroker, }); return (plugin: string): PluginEnvironment => { @@ -51,16 +49,38 @@ To allow connections from the frontend, you should also install the `@backstage/ Once you have both of the backend plugins installed, you can utilize the signal service by calling the `publish` method. This will publish the message to all subscribers in the frontend. To send message to -all subscribers, you can use `*` as `to` parameter. +all subscribers, you can use `null` as `recipients` parameter. ```ts // Periodic sending example setInterval(async () => { - await signalService.publish('*', 'plugin:topic', { - message: 'hello world', + await signalService.publish({ + recipients: null, + channel: 'my_plugin', + message: { + message: 'hello world', + }, }); }, 5000); ``` To receive this message in the frontend, check the documentation of `@backstage/plugin-signals` and `@backstage/plugin-signals-react`. + +## Using event broker directly + +Other way to send signals is to utilize the `EventBroker` directly. This requires that the payload is correct for it +to work: + +```ts +eventBroker.publish({ + topic: 'signals', + eventPayload: { + recipients: ['user:default/user1'], + message: { + message: 'hello world', + }, + channel: 'my_plugin', + }, +}); +``` diff --git a/plugins/signals-node/api-report.md b/plugins/signals-node/api-report.md index d85163040b..00d099f7db 100644 --- a/plugins/signals-node/api-report.md +++ b/plugins/signals-node/api-report.md @@ -3,63 +3,35 @@ > Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/). ```ts -/// - -import { Duplex } from 'stream'; import { EventBroker } from '@backstage/plugin-events-node'; -import http from 'http'; -import https from 'https'; -import { IdentityApi } from '@backstage/plugin-auth-node'; -import { IncomingMessage } from 'http'; import { JsonObject } from '@backstage/types'; -import { LoggerService } from '@backstage/backend-plugin-api'; import { ServiceRef } from '@backstage/backend-plugin-api'; // @public (undocumented) export class DefaultSignalService implements SignalService { // (undocumented) - static create(options: ServiceOptions): DefaultSignalService; - handleUpgrade(options: SignalServiceUpgradeOptions): Promise; - publish( - to: string | string[], - topic: string, - message: JsonObject, - ): Promise; + static create(options: SignalServiceOptions): DefaultSignalService; + publish(signal: SignalPayload): Promise; } // @public (undocumented) -export type ServiceOptions = { - eventBroker?: EventBroker; - logger: LoggerService; - identity: IdentityApi; -}; - -// @public (undocumented) -export type SignalEventBrokerPayload = { - recipients?: string[]; - topic?: string; - message?: JsonObject; +export type SignalPayload = { + recipients: string[] | null; + channel: string; + message: JsonObject; }; // @public (undocumented) export type SignalService = { - publish( - to: string | string[], - topic: string, - message: JsonObject, - ): Promise; - handleUpgrade(options: SignalServiceUpgradeOptions): Promise; + publish(signal: SignalPayload): Promise; }; // @public (undocumented) export const signalService: ServiceRef; // @public (undocumented) -export type SignalServiceUpgradeOptions = { - server: https.Server | http.Server; - request: IncomingMessage; - socket: Duplex; - head: Buffer; +export type SignalServiceOptions = { + eventBroker?: EventBroker; }; // (No @packageDocumentation comment for this package) diff --git a/plugins/signals-node/src/DefaultSignalService.ts b/plugins/signals-node/src/DefaultSignalService.ts index cab6e83c81..1fba96b8bc 100644 --- a/plugins/signals-node/src/DefaultSignalService.ts +++ b/plugins/signals-node/src/DefaultSignalService.ts @@ -13,226 +13,38 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import { EventBroker, EventParams } from '@backstage/plugin-events-node'; -import { - ServiceOptions, - SignalConnection, - SignalEventBrokerPayload, -} from './types'; -import { RawData, WebSocket, WebSocketServer } from 'ws'; -import { IncomingMessage } from 'http'; -import { v4 as uuid } from 'uuid'; -import { JsonObject } from '@backstage/types'; -import { - BackstageIdentityResponse, - IdentityApi, - IdentityApiGetIdentityRequest, -} from '@backstage/plugin-auth-node'; -import { LoggerService } from '@backstage/backend-plugin-api'; -import { SignalService, SignalServiceUpgradeOptions } from './SignalService'; +import { EventBroker } from '@backstage/plugin-events-node'; +import { SignalPayload, SignalServiceOptions } from './types'; +import { SignalService } from './SignalService'; /** @public */ export class DefaultSignalService implements SignalService { - private connections: Map = new Map< - string, - SignalConnection - >(); + // TODO: Remove this to be optional when events-backend has eventBroker as service private eventBroker?: EventBroker; - private logger: LoggerService; - private identity: IdentityApi; - private server: WebSocketServer; - static create(options: ServiceOptions) { + static create(options: SignalServiceOptions) { return new DefaultSignalService(options); } - private constructor(options: ServiceOptions) { - ({ - eventBroker: this.eventBroker, - logger: this.logger, - identity: this.identity, - } = options); - - this.server = new WebSocketServer({ - noServer: true, - clientTracking: false, - }); - - this.eventBroker?.subscribe({ - supportsEventTopics: () => ['signals'], - onEvent: (params: EventParams) => - this.onEventBrokerEvent(params), - }); - } - - /** - * Handles request upgrade to websocket and adds the connection to internal - * list for publish/subscribe functionality - * @param req - Request - */ - async handleUpgrade(options: SignalServiceUpgradeOptions) { - const { request, socket, head } = options; - let identity: BackstageIdentityResponse | undefined = undefined; - - // Authentication token is passed in Sec-WebSocket-Protocol header as there - // is no other way to pass the token with plain websockets - const token = request.headers['sec-websocket-protocol']; - if (token) { - identity = await this.identity.getIdentity({ - request: { - headers: { authorization: token }, - }, - } as IdentityApiGetIdentityRequest); - } - - this.server.handleUpgrade( - request, - socket, - head, - (ws: WebSocket, __: IncomingMessage) => { - this.addConnection(ws, identity); - }, - ); - } - - private addConnection(ws: WebSocket, identity?: BackstageIdentityResponse) { - const id = uuid(); - - const conn = { - id, - user: identity?.identity.userEntityRef ?? 'user:default/guest', - ws, - ownershipEntityRefs: identity?.identity.ownershipEntityRefs ?? [], - subscriptions: new Set(), - }; - - this.connections.set(id, conn); - - ws.on('error', (err: Error) => { - this.logger.info( - `Error occurred with connection ${id}: ${err}, closing connection`, - ); - ws.close(); - this.connections.delete(id); - }); - - ws.on('close', (code: number, reason: Buffer) => { - this.logger.info( - `Connection ${id} closed with code ${code}, reason: ${reason}`, - ); - this.connections.delete(id); - }); - - ws.on('message', (data: RawData, isBinary: boolean) => { - this.logger.debug(`Received message from connection ${id}: ${data}`); - if (isBinary) { - return; - } - try { - const json = JSON.parse(data.toString()) as JsonObject; - this.handleMessage(conn, json); - } catch (err: any) { - this.logger.error( - `Invalid message received from connection ${id}: ${err}`, - ); - } - }); - } - - private handleMessage(connection: SignalConnection, message: JsonObject) { - if (message.action === 'subscribe' && message.topic) { - this.logger.info( - `Connection ${connection.id} subscribed to ${message.topic}`, - ); - connection.subscriptions.add(message.topic as string); - } - - if (message.action === 'unsubscribe' && message.topic) { - this.logger.info( - `Connection ${connection.id} unsubscribed from ${message.topic}`, - ); - connection.subscriptions.delete(message.topic as string); - } + private constructor(options: SignalServiceOptions) { + ({ eventBroker: this.eventBroker } = options); } /** * Publishes a message to user refs to specific topic - * @param to - string or array of user ref strings to publish message to + * @param recipients - 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, - ); - } - - private async publishInternal( - recipients: string[], - topic: string, - message: JsonObject, - brokedEvent: boolean, - ) { - const jsonMessage = JSON.stringify({ topic, message }); - if (jsonMessage.length === 0) { - return; - } - - // If there is event broker, use that to publish the message to - // all signal services, including this one. - if (this.eventBroker && !brokedEvent) { - await this.eventBroker.publish({ - topic: 'signals', - eventPayload: { - recipients, - message, - topic, - }, - }); - return; - } - - // Actual websocket message sending - this.connections.forEach(conn => { - if (!conn.subscriptions.has(topic)) { - return; - } - // Sending to all users can be done with '*' - if ( - !recipients.includes('*') && - !conn.ownershipEntityRefs.some(ref => recipients.includes(ref)) - ) { - return; - } - - if (conn.ws.readyState !== WebSocket.OPEN) { - return; - } - - conn.ws.send(jsonMessage); + async publish(signal: SignalPayload) { + const { recipients, channel, message } = signal; + await this.eventBroker?.publish({ + topic: 'signals', + eventPayload: { + recipients, + message, + channel, + }, }); } - - private async onEventBrokerEvent( - params: EventParams, - ): Promise { - const { eventPayload } = params; - if ( - !eventPayload?.recipients || - !eventPayload.topic || - !eventPayload.message - ) { - return; - } - - await this.publishInternal( - eventPayload.recipients, - eventPayload.topic, - eventPayload.message, - true, - ); - } } diff --git a/plugins/signals-node/src/SignalService.ts b/plugins/signals-node/src/SignalService.ts index 7095c29858..f08a12661f 100644 --- a/plugins/signals-node/src/SignalService.ts +++ b/plugins/signals-node/src/SignalService.ts @@ -13,32 +13,12 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import { JsonObject } from '@backstage/types'; -import http, { IncomingMessage } from 'http'; -import { Duplex } from 'stream'; -import https from 'https'; - -/** @public */ -export type SignalServiceUpgradeOptions = { - server: https.Server | http.Server; - request: IncomingMessage; - socket: Duplex; - head: Buffer; -}; +import { SignalPayload } from './types'; /** @public */ export type SignalService = { /** * Publishes a message to user refs to specific topic */ - publish( - to: string | string[], - topic: string, - message: JsonObject, - ): Promise; - - /** - * Handles request upgrade - */ - handleUpgrade(options: SignalServiceUpgradeOptions): Promise; + publish(signal: SignalPayload): Promise; }; diff --git a/plugins/signals-node/src/lib.ts b/plugins/signals-node/src/lib.ts index c7e7c4a6ae..095a2f085d 100644 --- a/plugins/signals-node/src/lib.ts +++ b/plugins/signals-node/src/lib.ts @@ -14,7 +14,6 @@ * limitations under the License. */ import { - coreServices, createServiceFactory, createServiceRef, } from '@backstage/backend-plugin-api'; @@ -29,12 +28,11 @@ export const signalService = createServiceRef({ createServiceFactory({ service, deps: { - logger: coreServices.logger, - identity: coreServices.identity, - // TODO: EventBroker + // TODO: EventBroker. It is optional for now but it's actually required so waiting for the new backend system + // for the events-backend for this to work. }, - factory({ logger, identity }) { - return DefaultSignalService.create({ identity, logger }); + factory({}) { + return DefaultSignalService.create({}); }, }), }); diff --git a/plugins/signals-node/src/types.ts b/plugins/signals-node/src/types.ts index 6bfafae0b2..0220a196aa 100644 --- a/plugins/signals-node/src/types.ts +++ b/plugins/signals-node/src/types.ts @@ -13,35 +13,19 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import { IdentityApi } from '@backstage/plugin-auth-node'; import { EventBroker } from '@backstage/plugin-events-node'; -import { WebSocket } from 'ws'; import { JsonObject } from '@backstage/types'; -import { LoggerService } from '@backstage/backend-plugin-api'; /** * @public */ -export type ServiceOptions = { +export type SignalServiceOptions = { eventBroker?: EventBroker; - logger: LoggerService; - identity: IdentityApi; }; /** @public */ -export type SignalEventBrokerPayload = { - recipients?: string[]; - topic?: string; - message?: JsonObject; -}; - -/** - * @internal - */ -export type SignalConnection = { - id: string; - user: string; - ws: WebSocket; - ownershipEntityRefs: string[]; - subscriptions: Set; +export type SignalPayload = { + recipients: string[] | null; + channel: string; + message: JsonObject; }; diff --git a/plugins/signals-react/README.md b/plugins/signals-react/README.md index 244a4ec43d..4da9338866 100644 --- a/plugins/signals-react/README.md +++ b/plugins/signals-react/README.md @@ -22,10 +22,15 @@ Example of using the hook: ```ts import { useSignal } from '@backstage/plugin-signals-react'; -const { lastSignal } = useSignal('myplugin:topic'); +const { lastSignal } = useSignal('myplugin:channel'); + +useEffect(() => { + console.log(lastSignal); +}, [lastSignal]); ``` -Whenever backend publishes new message to the topic `myplugin:topic`, the lastSignal is changed. +Whenever backend publishes new message to the channel `myplugin:channel`, the `lastSignal` is changed. The `lastSignal` +is always initiated with null value before any messages are received from the backend. ## Using API directly @@ -37,7 +42,7 @@ import { signalsApiRef } from '@backstage/plugin-signals-react'; const signals = useApi(signalsApiRef); const { unsubscribe } = signals.subscribe( - 'myplugin:topic', + 'myplugin:channel', (message: JsonObject) => { console.log(message); }, diff --git a/plugins/signals-react/api-report.md b/plugins/signals-react/api-report.md index 8ad0c06fbe..2df2e4f1ce 100644 --- a/plugins/signals-react/api-report.md +++ b/plugins/signals-react/api-report.md @@ -9,7 +9,7 @@ import { JsonObject } from '@backstage/types'; // @public (undocumented) export type SignalApi = { subscribe( - topic: string, + channel: string, onMessage: (message: JsonObject) => void, ): { unsubscribe: () => void; @@ -20,7 +20,7 @@ export type SignalApi = { export const signalApiRef: ApiRef; // @public (undocumented) -export const useSignal: (topic: string) => { +export const useSignal: (channel: string) => { lastSignal: JsonObject | null; }; diff --git a/plugins/signals-react/src/api/SignalApi.ts b/plugins/signals-react/src/api/SignalApi.ts index 09eacafa4d..b37b3ae2f5 100644 --- a/plugins/signals-react/src/api/SignalApi.ts +++ b/plugins/signals-react/src/api/SignalApi.ts @@ -24,7 +24,7 @@ export const signalApiRef = createApiRef({ /** @public */ export type SignalApi = { subscribe( - topic: string, + channel: string, onMessage: (message: JsonObject) => void, ): { unsubscribe: () => void }; }; diff --git a/plugins/signals-react/src/hooks/useSignal.ts b/plugins/signals-react/src/hooks/useSignal.ts index 1176769a06..a7d50fbe3b 100644 --- a/plugins/signals-react/src/hooks/useSignal.ts +++ b/plugins/signals-react/src/hooks/useSignal.ts @@ -19,7 +19,7 @@ import { JsonObject } from '@backstage/types'; import { useEffect, useState } from 'react'; /** @public */ -export const useSignal = (topic: string) => { +export const useSignal = (channel: string) => { const apiHolder = useApiHolder(); // Use apiHolder instead useApi in case signalApi is not available in the // backstage instance this is used @@ -28,7 +28,7 @@ export const useSignal = (topic: string) => { useEffect(() => { let unsub: null | (() => void) = null; if (signals) { - const { unsubscribe } = signals.subscribe(topic, (msg: JsonObject) => { + const { unsubscribe } = signals.subscribe(channel, (msg: JsonObject) => { setLastSignal(msg); }); unsub = unsubscribe; @@ -38,7 +38,7 @@ export const useSignal = (topic: string) => { unsub(); } }; - }, [signals, topic]); + }, [signals, channel]); return { lastSignal }; }; diff --git a/plugins/signals/api-report.md b/plugins/signals/api-report.md index 97c882dda4..398d5fbfe6 100644 --- a/plugins/signals/api-report.md +++ b/plugins/signals/api-report.md @@ -24,7 +24,7 @@ export class SignalClient implements SignalApi { static readonly DEFAULT_RECONNECT_TIMEOUT_MS: number; // (undocumented) subscribe( - topic: string, + channel: string, onMessage: (message: JsonObject) => void, ): { unsubscribe: () => void; diff --git a/plugins/signals/src/api/SignalClient.ts b/plugins/signals/src/api/SignalClient.ts index 24a38a49e3..ee4ed8756b 100644 --- a/plugins/signals/src/api/SignalClient.ts +++ b/plugins/signals/src/api/SignalClient.ts @@ -19,7 +19,7 @@ import { DiscoveryApi, IdentityApi } from '@backstage/core-plugin-api'; import { v4 as uuid } from 'uuid'; type Subscription = { - topic: string; + channel: string; callback: (message: JsonObject) => void; }; @@ -63,20 +63,23 @@ export class SignalClient implements SignalApi { ) {} subscribe( - topic: string, + channel: string, onMessage: (message: JsonObject) => void, ): { unsubscribe: () => void } { const subscriptionId = uuid(); const exists = [...this.subscriptions.values()].find( - sub => sub.topic === topic, + sub => sub.channel === channel, ); - this.subscriptions.set(subscriptionId, { topic, callback: onMessage }); + this.subscriptions.set(subscriptionId, { + channel: channel, + callback: onMessage, + }); this.connect() .then(() => { - // Do not subscribe twice to same topic even there is multiple callbacks + // Do not subscribe twice to same channel even there is multiple callbacks if (!exists) { - this.send({ action: 'subscribe', topic }); + this.send({ action: 'subscribe', channel }); } }) .catch(() => { @@ -90,12 +93,12 @@ export class SignalClient implements SignalApi { } this.subscriptions.delete(subscriptionId); const multipleExists = [...this.subscriptions.values()].find( - s => s.topic === topic, + s => s.channel === channel, ); - // If there are subscriptions still listening to this topic, do not + // If there are subscriptions still listening to this channel, do not // unsubscribe from the server if (!multipleExists) { - this.send({ action: 'unsubscribe', topic: sub.topic }); + this.send({ action: 'unsubscribe', channel: sub.channel }); } // If there are no subscriptions, close the connection @@ -176,9 +179,9 @@ export class SignalClient implements SignalApi { private handleMessage(data: MessageEvent) { try { const json = JSON.parse(data.data) as JsonObject; - if (json.topic) { + if (json.channel) { for (const sub of this.subscriptions.values()) { - if (sub.topic === json.topic) { + if (sub.channel === json.channel) { sub.callback(json.message as JsonObject); } } @@ -201,9 +204,9 @@ export class SignalClient implements SignalApi { this.ws = null; this.connect() .then(() => { - // Resubscribe to existing topics in case we lost connection + // Resubscribe to existing channels in case we lost connection for (const sub of this.subscriptions.values()) { - this.send({ action: 'subscribe', topic: sub.topic }); + this.send({ action: 'subscribe', channel: sub.channel }); } }) .catch(() => { diff --git a/plugins/signals/src/api/SignalsClient.test.ts b/plugins/signals/src/api/SignalsClient.test.ts index ec202e2703..2e93cd69f1 100644 --- a/plugins/signals/src/api/SignalsClient.test.ts +++ b/plugins/signals/src/api/SignalsClient.test.ts @@ -44,21 +44,21 @@ describe('SignalsClient', () => { it('should handle single subscription correctly', async () => { const messageMock = jest.fn(); const client = SignalClient.create({ discoveryApi, identity }); - const { unsubscribe } = client.subscribe('topic', messageMock); + const { unsubscribe } = client.subscribe('channel', messageMock); await server.connected; await expect(server).toReceiveMessage({ action: 'subscribe', - topic: 'topic', + channel: 'channel', }); - server.send({ topic: 'topic', message: { hello: 'world' } }); + server.send({ channel: 'channel', message: { hello: 'world' } }); expect(messageMock).toHaveBeenCalledWith({ hello: 'world' }); await unsubscribe(); await expect(server).toReceiveMessage({ action: 'unsubscribe', - topic: 'topic', + channel: 'channel', }); }); @@ -68,11 +68,11 @@ describe('SignalsClient', () => { const client1 = SignalClient.create({ discoveryApi, identity }); const client2 = SignalClient.create({ discoveryApi, identity }); const { unsubscribe: unsubscribe1 } = client1.subscribe( - 'topic', + 'channel', messageMock1, ); const { unsubscribe: unsubscribe2 } = client2.subscribe( - 'topic', + 'channel', messageMock2, ); @@ -80,22 +80,22 @@ describe('SignalsClient', () => { await expect(server).toReceiveMessage({ action: 'subscribe', - topic: 'topic', + channel: 'channel', }); - server.send({ topic: 'topic', message: { hello: 'world' } }); + server.send({ channel: 'channel', message: { hello: 'world' } }); expect(messageMock1).toHaveBeenCalledWith({ hello: 'world' }); expect(messageMock2).toHaveBeenCalledWith({ hello: 'world' }); await unsubscribe1(); await expect(server).not.toReceiveMessage({ action: 'unsubscribe', - topic: 'topic', + channel: 'channel', }); await unsubscribe2(); await expect(server).toReceiveMessage({ action: 'unsubscribe', - topic: 'topic', + channel: 'channel', }); }); @@ -108,11 +108,11 @@ describe('SignalsClient', () => { connectTimeout: 100, }); - client.subscribe('topic', messageMock); + client.subscribe('channel', messageMock); await server.connected; await expect(server).toReceiveMessage({ action: 'subscribe', - topic: 'topic', + channel: 'channel', }); await server.server.emit('error', null); @@ -120,7 +120,7 @@ describe('SignalsClient', () => { await new Promise(r => setTimeout(r, 50)); await expect(server).toReceiveMessage({ action: 'subscribe', - topic: 'topic', + channel: 'channel', }); }); });