diff --git a/.changeset/poor-sheep-tease.md b/.changeset/poor-sheep-tease.md deleted file mode 100644 index fa89ac4749..0000000000 --- a/.changeset/poor-sheep-tease.md +++ /dev/null @@ -1,6 +0,0 @@ ---- -'@backstage/plugin-devtools-backend': patch -'@backstage/plugin-devtools': patch ---- - -Update devtools information almost real time using signals plugin diff --git a/packages/backend/src/index.ts b/packages/backend/src/index.ts index fc6b814c00..0e92698230 100644 --- a/packages/backend/src/index.ts +++ b/packages/backend/src/index.ts @@ -73,7 +73,7 @@ import { DefaultEventBroker } from '@backstage/plugin-events-backend'; import { PrometheusExporter } from '@opentelemetry/exporter-prometheus'; import { MeterProvider } from '@opentelemetry/sdk-metrics'; import { metrics } from '@opentelemetry/api'; -import { SignalService } from '@backstage/plugin-signals-node'; +import { DefaultSignalService } from '@backstage/plugin-signals-node'; // Expose opentelemetry metrics using a Prometheus exporter on // http://localhost:9464/metrics . See prometheus.yml in packages/backend for @@ -100,7 +100,7 @@ function makeCreateEnv(config: Config) { }); const eventBroker = new DefaultEventBroker(root.child({ type: 'plugin' })); - const signalService = SignalService.create({ + const signalService = DefaultSignalService.create({ logger: root, eventBroker, identity, diff --git a/packages/backend/src/plugins/devtools.ts b/packages/backend/src/plugins/devtools.ts index 5488033682..8e1767ddb1 100644 --- a/packages/backend/src/plugins/devtools.ts +++ b/packages/backend/src/plugins/devtools.ts @@ -25,6 +25,5 @@ export default async function createPlugin( logger: env.logger, config: env.config, permissions: env.permissions, - signalService: env.signalService, }); } diff --git a/packages/backend/src/types.ts b/packages/backend/src/types.ts index d76e68c1c9..3dad2f739b 100644 --- a/packages/backend/src/types.ts +++ b/packages/backend/src/types.ts @@ -27,7 +27,7 @@ import { PluginTaskScheduler } from '@backstage/backend-tasks'; import { IdentityApi } from '@backstage/plugin-auth-node'; import { PermissionEvaluator } from '@backstage/plugin-permission-common'; import { EventBroker } from '@backstage/plugin-events-node'; -import { SignalService } from '@backstage/plugin-signals-node'; +import { DefaultSignalService } from '@backstage/plugin-signals-node'; export type PluginEnvironment = { logger: Logger; @@ -41,5 +41,5 @@ export type PluginEnvironment = { scheduler: PluginTaskScheduler; identity: IdentityApi; eventBroker: EventBroker; - signalService: SignalService; + signalService: DefaultSignalService; }; diff --git a/plugins/devtools-backend/api-report.md b/plugins/devtools-backend/api-report.md index 7049285c02..b9eb7b5a0c 100644 --- a/plugins/devtools-backend/api-report.md +++ b/plugins/devtools-backend/api-report.md @@ -11,7 +11,6 @@ import express from 'express'; import { ExternalDependency } from '@backstage/plugin-devtools-common'; import { Logger } from 'winston'; import { PermissionEvaluator } from '@backstage/plugin-permission-common'; -import { SignalService } from '@backstage/plugin-signals-node'; // @public (undocumented) export function createRouter(options: RouterOptions): Promise; @@ -25,8 +24,6 @@ export class DevToolsBackendApi { listExternalDependencyDetails(): Promise; // (undocumented) listInfo(): Promise; - // (undocumented) - listResourceUtilization(): Promise; } // @public @@ -43,7 +40,5 @@ export interface RouterOptions { logger: Logger; // (undocumented) permissions: PermissionEvaluator; - // (undocumented) - signalService?: SignalService; } ``` diff --git a/plugins/devtools-backend/package.json b/plugins/devtools-backend/package.json index 8de7da1b39..18013bc52f 100644 --- a/plugins/devtools-backend/package.json +++ b/plugins/devtools-backend/package.json @@ -38,7 +38,6 @@ "@backstage/plugin-devtools-common": "workspace:^", "@backstage/plugin-permission-common": "workspace:^", "@backstage/plugin-permission-node": "workspace:^", - "@backstage/plugin-signals-node": "workspace:^", "@backstage/types": "workspace:^", "@manypkg/get-packages": "^1.1.3", "@types/express": "*", diff --git a/plugins/devtools-backend/src/api/DevToolsBackendApi.ts b/plugins/devtools-backend/src/api/DevToolsBackendApi.ts index da9a1ca0dd..b71edb3391 100644 --- a/plugins/devtools-backend/src/api/DevToolsBackendApi.ts +++ b/plugins/devtools-backend/src/api/DevToolsBackendApi.ts @@ -203,21 +203,17 @@ export class DevToolsBackendApi { return configInfo; } - public async listResourceUtilization(): Promise { + public async listInfo(): Promise { + const operatingSystem = `${os.hostname()}: ${os.type} ${os.release} - ${ + os.platform + }/${os.arch}`; const usedMem = Math.floor((os.totalmem() - os.freemem()) / (1024 * 1024)); - return `Memory: ${usedMem}/${Math.floor( + const resources = `Memory: ${usedMem}/${Math.floor( os.totalmem() / (1024 * 1024), )}MB - Load: ${os .loadavg() .map(v => v.toFixed(2)) .join('/')}`; - } - - public async listInfo(): Promise { - const operatingSystem = `${os.hostname()}: ${os.type} ${os.release} - ${ - os.platform - }/${os.arch}`; - const nodeJsVersion = process.version; /* eslint-disable-next-line no-restricted-syntax */ @@ -251,7 +247,7 @@ export class DevToolsBackendApi { const info: DevToolsInfo = { operatingSystem: operatingSystem ?? 'N/A', - resourceUtilization: (await this.listResourceUtilization()) ?? 'N/A', + resourceUtilization: resources ?? 'N/A', nodeJsVersion: nodeJsVersion ?? 'N/A', backstageVersion: backstageJson && backstageJson.version ? backstageJson.version : 'N/A', diff --git a/plugins/devtools-backend/src/plugin.ts b/plugins/devtools-backend/src/plugin.ts index 02c3831128..685522557e 100644 --- a/plugins/devtools-backend/src/plugin.ts +++ b/plugins/devtools-backend/src/plugin.ts @@ -20,7 +20,6 @@ import { createBackendPlugin, } from '@backstage/backend-plugin-api'; import { createRouter } from './service/router'; -import { signalService } from '@backstage/plugin-signals-node'; /** * DevTools backend plugin @@ -36,15 +35,13 @@ export const devtoolsPlugin = createBackendPlugin({ logger: coreServices.logger, permissions: coreServices.permissions, httpRouter: coreServices.httpRouter, - signals: signalService, }, - async init({ config, logger, permissions, httpRouter, signals }) { + async init({ config, logger, permissions, httpRouter }) { httpRouter.use( await createRouter({ config, logger: loggerToWinstonLogger(logger), permissions, - signalService: signals, }), ); }, diff --git a/plugins/devtools-backend/src/service/router.ts b/plugins/devtools-backend/src/service/router.ts index 106eb44309..fbeeb5af84 100644 --- a/plugins/devtools-backend/src/service/router.ts +++ b/plugins/devtools-backend/src/service/router.ts @@ -33,7 +33,6 @@ import { errorHandler } from '@backstage/backend-common'; import express from 'express'; import { getBearerTokenFromAuthorizationHeader } from '@backstage/plugin-auth-node'; import { createPermissionIntegrationRouter } from '@backstage/plugin-permission-node'; -import { SignalService } from '@backstage/plugin-signals-node'; /** @public */ export interface RouterOptions { @@ -41,30 +40,17 @@ export interface RouterOptions { logger: Logger; config: Config; permissions: PermissionEvaluator; - signalService?: SignalService; } /** @public */ export async function createRouter( options: RouterOptions, ): Promise { - const { logger, config, permissions, signalService } = options; + const { logger, config, permissions } = options; const devToolsBackendApi = options.devToolsBackendApi || new DevToolsBackendApi(logger, config); - if (signalService) { - // Publish info periodically using the signal service - setInterval(async () => { - if (signalService.hasSubscribers('devtools:resources')) { - const info = await devToolsBackendApi.listResourceUtilization(); - await signalService.publish('*', 'devtools:resources', { - resources: info, - }); - } - }, 5000); - } - const router = Router(); router.use(express.json()); router.use( diff --git a/plugins/devtools/src/components/Content/InfoContent/InfoContent.tsx b/plugins/devtools/src/components/Content/InfoContent/InfoContent.tsx index 2480bf5233..0ed43d2fee 100644 --- a/plugins/devtools/src/components/Content/InfoContent/InfoContent.tsx +++ b/plugins/devtools/src/components/Content/InfoContent/InfoContent.tsx @@ -29,7 +29,7 @@ import { Theme, } from '@material-ui/core'; import { Alert } from '@material-ui/lab'; -import React, { useEffect, useState } from 'react'; +import React from 'react'; import { useInfo } from '../../../hooks'; import { InfoDependenciesTable } from './InfoDependenciesTable'; import DescriptionIcon from '@material-ui/icons/Description'; @@ -38,7 +38,6 @@ import DeveloperBoardIcon from '@material-ui/icons/DeveloperBoard'; import { BackstageLogoIcon } from './BackstageLogoIcon'; import FileCopyIcon from '@material-ui/icons/FileCopy'; import { DevToolsInfo } from '@backstage/plugin-devtools-common'; -import { useSignalApi } from '@backstage/plugin-signals-react'; const useStyles = makeStyles((theme: Theme) => createStyles({ @@ -73,17 +72,7 @@ const copyToClipboard = ({ about }: { about: DevToolsInfo | undefined }) => { /** @public */ export const InfoContent = () => { const classes = useStyles(); - const [resources, setResources] = useState(undefined); const { about, loading, error } = useInfo(); - useSignalApi('devtools:resources', message => { - setResources(message.resources as string); - }); - - useEffect(() => { - if (!loading && !error && about) { - setResources(about.resourceUtilization); - } - }, [about, loading, error]); if (loading) { return ; @@ -113,7 +102,7 @@ export const InfoContent = () => { diff --git a/plugins/signals-backend/api-report.md b/plugins/signals-backend/api-report.md index afbc63abf2..e96dd2e09c 100644 --- a/plugins/signals-backend/api-report.md +++ b/plugins/signals-backend/api-report.md @@ -5,7 +5,7 @@ ```ts import { BackendFeature } from '@backstage/backend-plugin-api'; import express from 'express'; -import { Logger } from 'winston'; +import { LoggerService } from '@backstage/backend-plugin-api'; import { SignalService } from '@backstage/plugin-signals-node'; // @public (undocumented) @@ -14,7 +14,7 @@ export function createRouter(options: RouterOptions): Promise; // @public (undocumented) export interface RouterOptions { // (undocumented) - logger: Logger; + logger: LoggerService; // (undocumented) service: SignalService; } diff --git a/plugins/signals-backend/src/plugin.ts b/plugins/signals-backend/src/plugin.ts index 8ade4e85b5..68bbf27377 100644 --- a/plugins/signals-backend/src/plugin.ts +++ b/plugins/signals-backend/src/plugin.ts @@ -13,7 +13,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import { loggerToWinstonLogger } from '@backstage/backend-common'; import { coreServices, createBackendPlugin, @@ -38,7 +37,7 @@ export const signalsPlugin = createBackendPlugin({ async init({ httpRouter, logger, service }) { httpRouter.use( await createRouter({ - logger: loggerToWinstonLogger(logger), + logger, service, }), ); diff --git a/plugins/signals-backend/src/service/router.test.ts b/plugins/signals-backend/src/service/router.test.ts index b41704e008..281d6ed4a5 100644 --- a/plugins/signals-backend/src/service/router.test.ts +++ b/plugins/signals-backend/src/service/router.test.ts @@ -18,9 +18,9 @@ import express from 'express'; import request from 'supertest'; import { createRouter } from './router'; -import { SignalService } from '@backstage/plugin-signals-node'; +import { DefaultSignalService } from '@backstage/plugin-signals-node'; -const signalsServiceMock: jest.Mocked = {} as any; +const signalsServiceMock: jest.Mocked = {} as any; describe('createRouter', () => { let app: express.Express; diff --git a/plugins/signals-backend/src/service/router.ts b/plugins/signals-backend/src/service/router.ts index 8926910544..28bafce14a 100644 --- a/plugins/signals-backend/src/service/router.ts +++ b/plugins/signals-backend/src/service/router.ts @@ -16,14 +16,14 @@ import { errorHandler } from '@backstage/backend-common'; import express, { NextFunction, Request, Response } from 'express'; import Router from 'express-promise-router'; -import { Logger } from 'winston'; +import { LoggerService } from '@backstage/backend-plugin-api'; import { SignalService } from '@backstage/plugin-signals-node'; import * as https from 'https'; import http from 'http'; /** @public */ export interface RouterOptions { - logger: Logger; + logger: LoggerService; service: SignalService; } @@ -48,9 +48,9 @@ export async function createRouter( if (!subscribed) { subscribed = true; server.on('upgrade', async (request, socket, head) => { - // Only upgrade if request to root of the signals plugin + // TODO: Find a way to make this more generic if (request.url === '/api/signals') { - await service.handleUpgrade(request, socket, head); + await service.handleUpgrade({ server, request, socket, head }); } }); } diff --git a/plugins/signals-backend/src/service/standaloneServer.ts b/plugins/signals-backend/src/service/standaloneServer.ts index 0728b2fde5..e1ce27aee4 100644 --- a/plugins/signals-backend/src/service/standaloneServer.ts +++ b/plugins/signals-backend/src/service/standaloneServer.ts @@ -21,7 +21,7 @@ import { import { Server } from 'http'; import { Logger } from 'winston'; import { createRouter } from './router'; -import { SignalService } from '@backstage/plugin-signals-node'; +import { DefaultSignalService } from '@backstage/plugin-signals-node'; import { DefaultIdentityClient } from '@backstage/plugin-auth-node'; export interface ServerOptions { @@ -43,7 +43,7 @@ export async function startStandaloneServer( issuer: await discovery.getExternalBaseUrl('auth'), }); - const signals = SignalService.create({ + const signals = DefaultSignalService.create({ logger: logger, identity, }); diff --git a/plugins/signals-node/README.md b/plugins/signals-node/README.md index abea6c6d22..14c9332dc1 100644 --- a/plugins/signals-node/README.md +++ b/plugins/signals-node/README.md @@ -56,13 +56,9 @@ all subscribers, you can use `*` as `to` parameter. ```ts // Periodic sending example setInterval(async () => { - // You can use hasSubscribers to check if the message will be sent to anyone - // in case you need some heavy processing before that - if (signalService.hasSubscribers('plugin:topic')) { - await signalService.publish('*', 'plugin:topic', { - message: 'hello world', - }); - } + await signalService.publish('*', 'plugin:topic', { + message: 'hello world', + }); }, 5000); ``` diff --git a/plugins/signals-node/api-report.md b/plugins/signals-node/api-report.md index 7bfffc2540..d85163040b 100644 --- a/plugins/signals-node/api-report.md +++ b/plugins/signals-node/api-report.md @@ -7,14 +7,26 @@ import { Duplex } from 'stream'; import { EventBroker } from '@backstage/plugin-events-node'; -import { EventParams } from '@backstage/plugin-events-node'; -import { EventSubscriber } 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; +} + // @public (undocumented) export type ServiceOptions = { eventBroker?: EventBroker; @@ -30,28 +42,25 @@ export type SignalEventBrokerPayload = { }; // @public (undocumented) -export class SignalService implements EventSubscriber { - // (undocumented) - static create(options: ServiceOptions): SignalService; - handleUpgrade: ( - req: IncomingMessage, - socket: Duplex, - head: Buffer, - ) => Promise; - hasSubscribers(topic: string): boolean; - // (undocumented) - onEvent(params: EventParams): Promise; +export type SignalService = { publish( to: string | string[], topic: string, message: JsonObject, ): Promise; - // (undocumented) - supportsEventTopics(): string[]; -} + handleUpgrade(options: SignalServiceUpgradeOptions): Promise; +}; // @public (undocumented) export const signalService: ServiceRef; +// @public (undocumented) +export type SignalServiceUpgradeOptions = { + server: https.Server | http.Server; + request: IncomingMessage; + socket: Duplex; + head: Buffer; +}; + // (No @packageDocumentation comment for this package) ``` diff --git a/plugins/signals-node/package.json b/plugins/signals-node/package.json index d5e980d857..d3230432f1 100644 --- a/plugins/signals-node/package.json +++ b/plugins/signals-node/package.json @@ -37,7 +37,6 @@ "@backstage/types": "workspace:^", "express": "^4.17.1", "uuid": "^8.0.0", - "winston": "^3.2.1", "ws": "^8.14.2" } } diff --git a/plugins/signals-node/src/DefaultSignalService.ts b/plugins/signals-node/src/DefaultSignalService.ts new file mode 100644 index 0000000000..cab6e83c81 --- /dev/null +++ b/plugins/signals-node/src/DefaultSignalService.ts @@ -0,0 +1,238 @@ +/* + * 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 { + 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'; + +/** @public */ +export class DefaultSignalService implements SignalService { + private connections: Map = new Map< + string, + SignalConnection + >(); + private eventBroker?: EventBroker; + private logger: LoggerService; + private identity: IdentityApi; + private server: WebSocketServer; + + static create(options: ServiceOptions) { + 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); + } + } + + /** + * 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, + ); + } + + 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); + }); + } + + 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 9c621d5cbd..7095c29858 100644 --- a/plugins/signals-node/src/SignalService.ts +++ b/plugins/signals-node/src/SignalService.ts @@ -1,5 +1,5 @@ /* - * Copyright 2023 The Backstage Authors + * Copyright 2024 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. @@ -13,244 +13,32 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import { - EventBroker, - EventParams, - EventSubscriber, -} 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 http, { IncomingMessage } from 'http'; import { Duplex } from 'stream'; -import { LoggerService } from '@backstage/backend-plugin-api'; +import https from 'https'; /** @public */ -export class SignalService implements EventSubscriber { - private connections: Map = new Map< - string, - SignalConnection - >(); - private eventBroker?: EventBroker; - private logger: LoggerService; - private identity: IdentityApi; - private server: WebSocketServer; - - static create(options: ServiceOptions) { - return new SignalService(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(this); - } - - /** - * Handles request upgrade to websocket and adds the connection to internal - * list for publish/subscribe functionality - * @param req - Request - */ - handleUpgrade = async ( - req: IncomingMessage, - socket: Duplex, - head: Buffer, - ) => { - 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 = req.headers['sec-websocket-protocol']; - if (token) { - identity = await this.identity.getIdentity({ - request: { - headers: { authorization: token }, - }, - } as IdentityApiGetIdentityRequest); - } - - this.server.handleUpgrade( - req, - 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); - } - } +export type SignalServiceUpgradeOptions = { + server: https.Server | http.Server; + request: IncomingMessage; + socket: Duplex; + head: Buffer; +}; +/** @public */ +export type SignalService = { /** * 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, - ); - } - - /** - * Checks if there is active subscriptions to specific topic. - * This can be useful to skip heavy processing before publishing messages if there are no subscriptions. - * @param topic - topic to check for subscriptions - */ - hasSubscribers(topic: string): boolean { - return ( - [...this.connections.values()].find(conn => - conn.subscriptions.has(topic), - ) !== undefined - ); - } - - private async publishInternal( - recipients: string[], + publish( + to: string | string[], topic: string, message: JsonObject, - brokedEvent: boolean, - ) { - const jsonMessage = JSON.stringify({ topic, message }); - if (jsonMessage.length === 0) { - return; - } + ): Promise; - // 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 onEvent(params: EventParams): Promise { - const { eventPayload } = params; - if ( - !eventPayload?.recipients || - !eventPayload.topic || - !eventPayload.message - ) { - return; - } - - await this.publishInternal( - eventPayload.recipients, - eventPayload.topic, - eventPayload.message, - true, - ); - } - - supportsEventTopics(): string[] { - return ['signals']; - } -} + /** + * Handles request upgrade + */ + handleUpgrade(options: SignalServiceUpgradeOptions): Promise; +}; diff --git a/plugins/signals-node/src/index.ts b/plugins/signals-node/src/index.ts index 43a3a88ac0..f9c220b1e9 100644 --- a/plugins/signals-node/src/index.ts +++ b/plugins/signals-node/src/index.ts @@ -15,5 +15,6 @@ */ export * from './lib'; +export * from './DefaultSignalService'; export * from './SignalService'; export * from './types'; diff --git a/plugins/signals-node/src/lib.ts b/plugins/signals-node/src/lib.ts index b95d4a9231..c7e7c4a6ae 100644 --- a/plugins/signals-node/src/lib.ts +++ b/plugins/signals-node/src/lib.ts @@ -18,6 +18,7 @@ import { createServiceFactory, createServiceRef, } from '@backstage/backend-plugin-api'; +import { DefaultSignalService } from './DefaultSignalService'; import { SignalService } from './SignalService'; /** @public */ @@ -33,7 +34,7 @@ export const signalService = createServiceRef({ // TODO: EventBroker }, factory({ logger, identity }) { - return SignalService.create({ identity, logger }); + return DefaultSignalService.create({ identity, logger }); }, }), }); diff --git a/plugins/signals-react/README.md b/plugins/signals-react/README.md index 9bdf5e964f..244a4ec43d 100644 --- a/plugins/signals-react/README.md +++ b/plugins/signals-react/README.md @@ -20,15 +20,12 @@ of connections to the backend and also to allow multiple subscriptions using the Example of using the hook: ```ts -import { useSignalApi } from '@backstage/plugin-signals-react'; +import { useSignal } from '@backstage/plugin-signals-react'; -const [message, setMessage] = useState(undefined); -useSignalApi('myplugin:topic', message => { - setMessage(message); -}); +const { lastSignal } = useSignal('myplugin:topic'); ``` -Whenever backend publishes new message to the topic `myplugin:topic`, the state of this component is changed. +Whenever backend publishes new message to the topic `myplugin:topic`, the lastSignal is changed. ## Using API directly @@ -39,9 +36,12 @@ subscriptions. import { signalsApiRef } from '@backstage/plugin-signals-react'; const signals = useApi(signalsApiRef); -signals.subscribe('myplugin:topic', (message: JsonObject) => { - console.log(message); -}); +const { unsubscribe } = signals.subscribe( + 'myplugin:topic', + (message: JsonObject) => { + console.log(message); + }, +); // Remember to unsubscribe -signals.unsubscribe('myplugin:topic'); +unsubscribe(); ``` diff --git a/plugins/signals-react/api-report.md b/plugins/signals-react/api-report.md index d28b806103..8ad0c06fbe 100644 --- a/plugins/signals-react/api-report.md +++ b/plugins/signals-react/api-report.md @@ -8,18 +8,21 @@ import { JsonObject } from '@backstage/types'; // @public (undocumented) export type SignalApi = { - subscribe(topic: string, onMessage: (message: JsonObject) => void): string; - unsubscribe(subscription: string): void; + subscribe( + topic: string, + onMessage: (message: JsonObject) => void, + ): { + unsubscribe: () => void; + }; }; // @public (undocumented) export const signalApiRef: ApiRef; // @public (undocumented) -export const useSignalApi: ( - topic: string, - onMessage: (message: JsonObject) => void, -) => void; +export const useSignal: (topic: string) => { + lastSignal: JsonObject | null; +}; // (No @packageDocumentation comment for this package) ``` diff --git a/plugins/signals-react/package.json b/plugins/signals-react/package.json index 148bc27cb3..06df050ccd 100644 --- a/plugins/signals-react/package.json +++ b/plugins/signals-react/package.json @@ -26,7 +26,7 @@ "dependencies": { "@backstage/core-plugin-api": "workspace:^", "@backstage/types": "workspace:^", - "@material-ui/core": "^4.9.13" + "@material-ui/core": "^4.12.4" }, "peerDependencies": { "react": "^16.13.1 || ^17.0.0" @@ -34,8 +34,8 @@ "devDependencies": { "@backstage/cli": "workspace:^", "@backstage/test-utils": "workspace:^", - "@testing-library/jest-dom": "^5.10.1", - "@testing-library/react": "^12.1.3" + "@testing-library/jest-dom": "^6.0.0", + "@testing-library/react": "^14.0.0" }, "files": [ "dist" diff --git a/plugins/signals-react/src/api/SignalApi.ts b/plugins/signals-react/src/api/SignalApi.ts index f63023bf5f..09eacafa4d 100644 --- a/plugins/signals-react/src/api/SignalApi.ts +++ b/plugins/signals-react/src/api/SignalApi.ts @@ -23,7 +23,8 @@ export const signalApiRef = createApiRef({ /** @public */ export type SignalApi = { - subscribe(topic: string, onMessage: (message: JsonObject) => void): string; - - unsubscribe(subscription: string): void; + subscribe( + topic: string, + onMessage: (message: JsonObject) => void, + ): { unsubscribe: () => void }; }; diff --git a/plugins/signals-react/src/hooks/index.ts b/plugins/signals-react/src/hooks/index.ts index 921eca8d30..a7bf7ebad2 100644 --- a/plugins/signals-react/src/hooks/index.ts +++ b/plugins/signals-react/src/hooks/index.ts @@ -13,4 +13,4 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -export * from './useSignalApi'; +export * from './useSignal'; diff --git a/plugins/signals-react/src/hooks/useSignalApi.ts b/plugins/signals-react/src/hooks/useSignal.ts similarity index 69% rename from plugins/signals-react/src/hooks/useSignalApi.ts rename to plugins/signals-react/src/hooks/useSignal.ts index 8534732264..1176769a06 100644 --- a/plugins/signals-react/src/hooks/useSignalApi.ts +++ b/plugins/signals-react/src/hooks/useSignal.ts @@ -19,27 +19,26 @@ import { JsonObject } from '@backstage/types'; import { useEffect, useState } from 'react'; /** @public */ -export const useSignalApi = ( - topic: string, - onMessage: (message: JsonObject) => void, -) => { +export const useSignal = (topic: string) => { const apiHolder = useApiHolder(); // Use apiHolder instead useApi in case signalApi is not available in the // backstage instance this is used const signals = apiHolder.get(signalApiRef); - const [subscription, setSubscription] = useState(null); + const [lastSignal, setLastSignal] = useState(null); useEffect(() => { - if (signals && !subscription) { - const sub = signals.subscribe(topic, onMessage); - setSubscription(sub); + let unsub: null | (() => void) = null; + if (signals) { + const { unsubscribe } = signals.subscribe(topic, (msg: JsonObject) => { + setLastSignal(msg); + }); + unsub = unsubscribe; } - }, [subscription, signals, onMessage, topic]); - - useEffect(() => { return () => { - if (signals && subscription) { - signals.unsubscribe(subscription); + if (signals && unsub) { + unsub(); } }; - }, [subscription, signals, topic]); + }, [signals, topic]); + + return { lastSignal }; }; diff --git a/plugins/signals/api-report.md b/plugins/signals/api-report.md index 28a70a339c..97c882dda4 100644 --- a/plugins/signals/api-report.md +++ b/plugins/signals/api-report.md @@ -23,9 +23,12 @@ export class SignalClient implements SignalApi { // (undocumented) static readonly DEFAULT_RECONNECT_TIMEOUT_MS: number; // (undocumented) - subscribe(topic: string, onMessage: (message: JsonObject) => void): string; - // (undocumented) - unsubscribe(subscription: string): void; + subscribe( + topic: string, + onMessage: (message: JsonObject) => void, + ): { + unsubscribe: () => void; + }; } // @public (undocumented) diff --git a/plugins/signals/package.json b/plugins/signals/package.json index e708d46329..cec95280c1 100644 --- a/plugins/signals/package.json +++ b/plugins/signals/package.json @@ -28,7 +28,7 @@ "@backstage/plugin-signals-react": "workspace:^", "@backstage/theme": "workspace:^", "@backstage/types": "workspace:^", - "@material-ui/core": "^4.9.13", + "@material-ui/core": "^4.12.4", "@material-ui/icons": "^4.9.1", "@material-ui/lab": "^4.0.0-alpha.61", "react-use": "^17.2.4", @@ -42,8 +42,8 @@ "@backstage/core-app-api": "workspace:^", "@backstage/dev-utils": "workspace:^", "@backstage/test-utils": "workspace:^", - "@testing-library/jest-dom": "^5.10.1", - "@testing-library/react": "^12.1.3", + "@testing-library/jest-dom": "^6.0.0", + "@testing-library/react": "^14.0.0", "@testing-library/user-event": "^14.0.0", "jest-websocket-mock": "^2.5.0", "msw": "^1.0.0" diff --git a/plugins/signals/src/api/SignalClient.ts b/plugins/signals/src/api/SignalClient.ts index 64c1866fab..24a38a49e3 100644 --- a/plugins/signals/src/api/SignalClient.ts +++ b/plugins/signals/src/api/SignalClient.ts @@ -62,7 +62,10 @@ export class SignalClient implements SignalApi { private reconnectTimeout: number, ) {} - subscribe(topic: string, onMessage: (message: JsonObject) => void): string { + subscribe( + topic: string, + onMessage: (message: JsonObject) => void, + ): { unsubscribe: () => void } { const subscriptionId = uuid(); const exists = [...this.subscriptions.values()].find( sub => sub.topic === topic, @@ -79,30 +82,30 @@ export class SignalClient implements SignalApi { .catch(() => { this.reconnect(); }); - return subscriptionId; - } - unsubscribe(subscription: string): void { - const sub = this.subscriptions.get(subscription); - if (!sub) { - return; - } - const topic = sub.topic; - this.subscriptions.delete(subscription); - const exists = [...this.subscriptions.values()].find( - s => s.topic === topic, - ); - // If there are subscriptions still listening to this topic, do not - // unsubscribe from the server - if (!exists) { - this.send({ action: 'unsubscribe', topic: sub.topic }); - } + const unsubscribe = () => { + const sub = this.subscriptions.get(subscriptionId); + if (!sub) { + return; + } + this.subscriptions.delete(subscriptionId); + const multipleExists = [...this.subscriptions.values()].find( + s => s.topic === topic, + ); + // If there are subscriptions still listening to this topic, do not + // unsubscribe from the server + if (!multipleExists) { + this.send({ action: 'unsubscribe', topic: sub.topic }); + } - // If there are no subscriptions, close the connection - if (this.subscriptions.size === 0) { - this.ws?.close(WS_CLOSE_NORMAL); - this.ws = null; - } + // If there are no subscriptions, close the connection + if (this.subscriptions.size === 0) { + this.ws?.close(WS_CLOSE_NORMAL); + this.ws = null; + } + }; + + return { unsubscribe }; } private send(data?: JsonObject): void { diff --git a/plugins/signals/src/api/SignalsClient.test.ts b/plugins/signals/src/api/SignalsClient.test.ts index 60d67e6c4b..ec202e2703 100644 --- a/plugins/signals/src/api/SignalsClient.test.ts +++ b/plugins/signals/src/api/SignalsClient.test.ts @@ -44,7 +44,7 @@ describe('SignalsClient', () => { it('should handle single subscription correctly', async () => { const messageMock = jest.fn(); const client = SignalClient.create({ discoveryApi, identity }); - const sub = client.subscribe('topic', messageMock); + const { unsubscribe } = client.subscribe('topic', messageMock); await server.connected; await expect(server).toReceiveMessage({ @@ -54,7 +54,8 @@ describe('SignalsClient', () => { server.send({ topic: 'topic', message: { hello: 'world' } }); expect(messageMock).toHaveBeenCalledWith({ hello: 'world' }); - client.unsubscribe(sub); + await unsubscribe(); + await expect(server).toReceiveMessage({ action: 'unsubscribe', topic: 'topic', @@ -66,8 +67,14 @@ describe('SignalsClient', () => { const messageMock2 = jest.fn(); const client1 = SignalClient.create({ discoveryApi, identity }); const client2 = SignalClient.create({ discoveryApi, identity }); - const sub1 = client1.subscribe('topic', messageMock1); - const sub2 = client2.subscribe('topic', messageMock2); + const { unsubscribe: unsubscribe1 } = client1.subscribe( + 'topic', + messageMock1, + ); + const { unsubscribe: unsubscribe2 } = client2.subscribe( + 'topic', + messageMock2, + ); await server.connected; @@ -79,13 +86,13 @@ describe('SignalsClient', () => { expect(messageMock1).toHaveBeenCalledWith({ hello: 'world' }); expect(messageMock2).toHaveBeenCalledWith({ hello: 'world' }); - client1.unsubscribe(sub1); + await unsubscribe1(); await expect(server).not.toReceiveMessage({ action: 'unsubscribe', topic: 'topic', }); - client2.unsubscribe(sub2); + await unsubscribe2(); await expect(server).toReceiveMessage({ action: 'unsubscribe', topic: 'topic', diff --git a/plugins/signals/src/plugin.ts b/plugins/signals/src/plugin.ts index 1482d7dae1..eb3f9126e2 100644 --- a/plugins/signals/src/plugin.ts +++ b/plugins/signals/src/plugin.ts @@ -22,8 +22,6 @@ import { import { signalApiRef } from '@backstage/plugin-signals-react'; import { SignalClient } from './api/SignalClient'; -let clientInstance: SignalClient | undefined = undefined; - /** @public */ export const signalsPlugin = createPlugin({ id: 'signals', @@ -35,13 +33,10 @@ export const signalsPlugin = createPlugin({ discoveryApi: discoveryApiRef, }, factory: ({ identity, discoveryApi }) => { - if (!clientInstance) { - clientInstance = SignalClient.create({ - identity, - discoveryApi, - }); - } - return clientInstance; + return SignalClient.create({ + identity, + discoveryApi, + }); }, }), ], diff --git a/yarn.lock b/yarn.lock index 7a3fbb3902..72e4e43d02 100644 --- a/yarn.lock +++ b/yarn.lock @@ -12,7 +12,7 @@ __metadata: languageName: node linkType: hard -"@adobe/css-tools@npm:^4.0.1, @adobe/css-tools@npm:^4.3.2": +"@adobe/css-tools@npm:^4.3.2": version: 4.3.2 resolution: "@adobe/css-tools@npm:4.3.2" checksum: 9667d61d55dc3b0a315c530ae84e016ce5267c4dd8ac00abb40108dc98e07b98e3090ce8b87acd51a41a68d9e84dcccb08cdf21c902572a9cf9dcaf830da4ae3 @@ -6203,7 +6203,6 @@ __metadata: "@backstage/plugin-devtools-common": "workspace:^" "@backstage/plugin-permission-common": "workspace:^" "@backstage/plugin-permission-node": "workspace:^" - "@backstage/plugin-signals-node": "workspace:^" "@backstage/types": "workspace:^" "@manypkg/get-packages": ^1.1.3 "@types/express": "*" @@ -8875,12 +8874,11 @@ __metadata: "@types/express": ^4.17.21 express: ^4.17.1 uuid: ^8.0.0 - winston: ^3.2.1 ws: ^8.14.2 languageName: unknown linkType: soft -"@backstage/plugin-signals-react@workspace:plugins/signals-react": +"@backstage/plugin-signals-react@workspace:^, @backstage/plugin-signals-react@workspace:plugins/signals-react": version: 0.0.0-use.local resolution: "@backstage/plugin-signals-react@workspace:plugins/signals-react" dependencies: @@ -8888,9 +8886,9 @@ __metadata: "@backstage/core-plugin-api": "workspace:^" "@backstage/test-utils": "workspace:^" "@backstage/types": "workspace:^" - "@material-ui/core": ^4.9.13 - "@testing-library/jest-dom": ^5.10.1 - "@testing-library/react": ^12.1.3 + "@material-ui/core": ^4.12.4 + "@testing-library/jest-dom": ^6.0.0 + "@testing-library/react": ^14.0.0 peerDependencies: react: ^16.13.1 || ^17.0.0 languageName: unknown @@ -8909,11 +8907,11 @@ __metadata: "@backstage/test-utils": "workspace:^" "@backstage/theme": "workspace:^" "@backstage/types": "workspace:^" - "@material-ui/core": ^4.9.13 + "@material-ui/core": ^4.12.4 "@material-ui/icons": ^4.9.1 "@material-ui/lab": ^4.0.0-alpha.61 - "@testing-library/jest-dom": ^5.10.1 - "@testing-library/react": ^12.1.3 + "@testing-library/jest-dom": ^6.0.0 + "@testing-library/react": ^14.0.0 "@testing-library/user-event": ^14.0.0 jest-websocket-mock: ^2.5.0 msw: ^1.0.0 @@ -17216,22 +17214,6 @@ __metadata: languageName: unknown linkType: soft -"@testing-library/dom@npm:^8.0.0": - version: 8.20.1 - resolution: "@testing-library/dom@npm:8.20.1" - dependencies: - "@babel/code-frame": ^7.10.4 - "@babel/runtime": ^7.12.5 - "@types/aria-query": ^5.0.1 - aria-query: 5.1.3 - chalk: ^4.1.0 - dom-accessibility-api: ^0.5.9 - lz-string: ^1.5.0 - pretty-format: ^27.0.2 - checksum: 06fc8dc67849aadb726cbbad0e7546afdf8923bd39acb64c576d706249bd7d0d05f08e08a31913fb621162e3b9c2bd0dce15964437f030f9fa4476326fdd3007 - languageName: node - linkType: hard - "@testing-library/dom@npm:^9.0.0": version: 9.3.3 resolution: "@testing-library/dom@npm:9.3.3" @@ -17248,23 +17230,6 @@ __metadata: languageName: node linkType: hard -"@testing-library/jest-dom@npm:^5.10.1": - version: 5.17.0 - resolution: "@testing-library/jest-dom@npm:5.17.0" - dependencies: - "@adobe/css-tools": ^4.0.1 - "@babel/runtime": ^7.9.2 - "@types/testing-library__jest-dom": ^5.9.1 - aria-query: ^5.0.0 - chalk: ^3.0.0 - css.escape: ^1.5.1 - dom-accessibility-api: ^0.5.6 - lodash: ^4.17.15 - redent: ^3.0.0 - checksum: 9f28dbca8b50d7c306aae40c3aa8e06f0e115f740360004bd87d57f95acf7ab4b4f4122a7399a76dbf2bdaaafb15c99cc137fdcb0ae457a92e2de0f3fbf9b03b - languageName: node - linkType: hard - "@testing-library/jest-dom@npm:^6.0.0": version: 6.1.6 resolution: "@testing-library/jest-dom@npm:6.1.6" @@ -17317,20 +17282,6 @@ __metadata: languageName: node linkType: hard -"@testing-library/react@npm:^12.1.3": - version: 12.1.5 - resolution: "@testing-library/react@npm:12.1.5" - dependencies: - "@babel/runtime": ^7.12.5 - "@testing-library/dom": ^8.0.0 - "@types/react-dom": <18.0.0 - peerDependencies: - react: <18.0.0 - react-dom: <18.0.0 - checksum: 4abd0490405e709a7df584a0db604e508a4612398bb1326e8fa32dd9393b15badc826dcf6d2f7525437886d507871f719f127b9860ed69ddd204d1fa834f576a - languageName: node - linkType: hard - "@testing-library/react@npm:^14.0.0": version: 14.1.2 resolution: "@testing-library/react@npm:14.1.2"