tweak exports to allow decoupling of backend plugins

Signed-off-by: Brian Fletcher <brian@roadie.io>

.changeset/tall-meals-dress.md

@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-events-backend': patch
+---
+
+Export `InMemoryEventBroker` to allow decoupling of the catalog and events backends in the `example-backend`.

packages/backend/src/index.ts

@@ -42,7 +42,6 @@ import { metricsInit, metricsHandler } from './metrics';
 import auth from './plugins/auth';
 import azureDevOps from './plugins/azure-devops';
 import catalog from './plugins/catalog';
-import catalogEventBasedProviders from './plugins/catalogEventBasedProviders';
 import codeCoverage from './plugins/codecoverage';
 import entityFeedback from './plugins/entityFeedback';
 import events from './plugins/events';
@@ -68,6 +67,7 @@ import linguist from './plugins/linguist';
 import { PluginEnvironment } from './types';
 import { ServerPermissionClient } from '@backstage/plugin-permission-node';
 import { DefaultIdentityClient } from '@backstage/plugin-auth-node';
+import { InMemoryEventBroker } from '@backstage/plugin-events-backend';
 
 function makeCreateEnv(config: Config) {
   const root = getRootLogger();
@@ -85,6 +85,8 @@ function makeCreateEnv(config: Config) {
     discovery,
   });
 
+  const eventBroker = new InMemoryEventBroker(root.child({ type: 'plugin' }));
+
   root.info(`Created UrlReader ${reader}`);
 
   return (plugin: string): PluginEnvironment => {
@@ -99,6 +101,7 @@ function makeCreateEnv(config: Config) {
       database,
       config,
       reader,
+      eventBroker,
       discovery,
       tokenManager,
       permissions,
@@ -156,18 +159,12 @@ async function main() {
   const exploreEnv = useHotMemoize(module, () => createEnv('explore'));
   const lighthouseEnv = useHotMemoize(module, () => createEnv('lighthouse'));
 
-  const eventBasedEntityProviders = await catalogEventBasedProviders(
-    catalogEnv,
-  );
   const linguistEnv = useHotMemoize(module, () => createEnv('linguist'));
 
   const apiRouter = Router();
-  apiRouter.use(
-    '/catalog',
-    await catalog(catalogEnv, eventBasedEntityProviders),
-  );
+  apiRouter.use('/catalog', await catalog(catalogEnv));
   apiRouter.use('/code-coverage', await codeCoverage(codeCoverageEnv));
-  apiRouter.use('/events', await events(eventsEnv, eventBasedEntityProviders));
+  apiRouter.use('/events', await events(eventsEnv));
   apiRouter.use('/rollbar', await rollbar(rollbarEnv));
   apiRouter.use('/scaffolder', await scaffolder(scaffolderEnv));
   apiRouter.use('/tech-insights', await techInsights(techInsightsEnv));

packages/backend/src/plugins/catalog.ts

@@ -15,18 +15,19 @@
  */
 
 import { CatalogBuilder } from '@backstage/plugin-catalog-backend';
-import { EntityProvider } from '@backstage/plugin-catalog-node';
 import { ScaffolderEntitiesProcessor } from '@backstage/plugin-scaffolder-backend';
 import { Router } from 'express';
 import { PluginEnvironment } from '../types';
+import createCatalogEventBasedProviders from './catalogEventBasedProviders';
 
 export default async function createPlugin(
   env: PluginEnvironment,
-  providers?: Array<EntityProvider>,
 ): Promise<Router> {
+  const providers = await createCatalogEventBasedProviders(env);
   const builder = await CatalogBuilder.create(env);
   builder.addProcessor(new ScaffolderEntitiesProcessor());
-  builder.addEntityProvider(providers ?? []);
+  env.eventBroker.subscribe(providers);
+  builder.addEntityProvider(providers);
   const { processingEngine, router } = await builder.build();
   await processingEngine.start();
   return router;

packages/backend/src/plugins/catalogEventBasedProviders.ts

@@ -29,10 +29,11 @@ class DemoEventBasedEntityProvider implements EntityProvider, EventSubscriber {
   ) {}
 
   async onEvent(params: EventParams): Promise<void> {
+    const request = params.eventPayload as Request;
     this.logger.info(
       `onEvent: topic=${params.topic}, metadata=${JSON.stringify(
         params.metadata,
-      )}, payload=${JSON.stringify(params.eventPayload)}`,
+      )}, payload=${JSON.stringify(request.body)}`,
     );
   }
 

packages/backend/src/plugins/events.ts

@@ -18,13 +18,11 @@ import {
   EventsBackend,
   HttpPostIngressEventPublisher,
 } from '@backstage/plugin-events-backend';
-import { EventSubscriber } from '@backstage/plugin-events-node';
 import { Router } from 'express';
 import { PluginEnvironment } from '../types';
 
 export default async function createPlugin(
   env: PluginEnvironment,
-  subscribers: EventSubscriber[],
 ): Promise<Router> {
   const eventsRouter = Router();
 
@@ -35,8 +33,8 @@ export default async function createPlugin(
   http.bind(eventsRouter);
 
   await new EventsBackend(env.logger)
+    .setEventBroker(env.eventBroker)
     .addPublishers(http)
-    .addSubscribers(subscribers)
     .start();
 
   return eventsRouter;

packages/backend/src/types.ts

@@ -26,6 +26,7 @@ import {
 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';
 
 export type PluginEnvironment = {
   logger: Logger;
@@ -38,4 +39,5 @@ export type PluginEnvironment = {
   permissions: PermissionEvaluator;
   scheduler: PluginTaskScheduler;
   identity: IdentityApi;
+  eventBroker: EventBroker;
 };

plugins/events-backend/api-report.md

@@ -5,6 +5,7 @@
 ```ts
 import { Config } from '@backstage/config';
 import { EventBroker } from '@backstage/plugin-events-node';
+import { EventParams } from '@backstage/plugin-events-node';
 import { EventPublisher } from '@backstage/plugin-events-node';
 import { EventSubscriber } from '@backstage/plugin-events-node';
 import express from 'express';
@@ -42,4 +43,15 @@ export class HttpPostIngressEventPublisher implements EventPublisher {
   // (undocumented)
   setEventBroker(eventBroker: EventBroker): Promise<void>;
 }
+
+// @public
+export class InMemoryEventBroker implements EventBroker {
+  constructor(logger: Logger);
+  // (undocumented)
+  publish(params: EventParams): Promise<void>;
+  // (undocumented)
+  subscribe(
+    ...subscribers: Array<EventSubscriber | Array<EventSubscriber>>
+  ): void;
+}
 ```

plugins/events-backend/src/index.ts

@@ -22,3 +22,4 @@
 
 export { EventsBackend } from './service/EventsBackend';
 export { HttpPostIngressEventPublisher } from './service/http';
+export { InMemoryEventBroker } from './service/InMemoryEventBroker';

plugins/events-backend/src/service/InMemoryEventBroker.ts

@@ -25,6 +25,8 @@ import { Logger } from 'winston';
  * In-memory event broker which will pass the event to all registered subscribers
  * interested in it.
  * Events will not be persisted in any form.
+ *
+ * @public
  */
 // TODO(pjungermann): add prom metrics? (see plugins/catalog-backend/src/util/metrics.ts, etc.)
 export class InMemoryEventBroker implements EventBroker {

plugins/events-backend/src/service/http/HttpPostIngressEventPublisher.ts

@@ -107,7 +107,7 @@ export class HttpPostIngressEventPublisher implements EventPublisher {
         return;
       }
 
-      const eventPayload = request.body;
+      const eventPayload = request;
       await this.eventBroker!.publish({
         topic,
         eventPayload,