Merge pull request #27763 from backstage/blam/incremental-events

incremental: wire up the events system in new backend

.changeset/rich-beers-eat.md

@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-catalog-backend-module-incremental-ingestion': patch
+---
+
+Wire up the events together in the new backend system

plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.test.ts

@@ -68,6 +68,7 @@ describe('WrapperProviders', () => {
         client,
         scheduler: scheduler as Partial<SchedulerService> as SchedulerService,
         applyDatabaseMigrations,
+        events: mockServices.events.mock(),
       });
       const wrapped1 = providers.wrap(provider1, {
         burstInterval: { seconds: 1 },

plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts

@@ -36,6 +36,7 @@ import {
   IncrementalEntityProvider,
   IncrementalEntityProviderOptions,
 } from '../types';
+import { EventsService } from '@backstage/plugin-events-node';
 
 /**
  * Helps in the creation of the catalog entity providers that wrap the
@@ -53,6 +54,7 @@ export class WrapperProviders {
       client: Knex;
       scheduler: SchedulerService;
       applyDatabaseMigrations?: typeof applyDatabaseMigrations;
+      events: EventsService;
     },
   ) {}
 
@@ -130,6 +132,20 @@ export class WrapperProviders {
         frequency,
         timeout: length,
       });
+
+      const topics = engine.supportsEventTopics();
+      if (topics.length > 0) {
+        logger.info(
+          `Provider ${provider.getProviderName()} subscribing to events for topics: ${topics.join(
+            ',',
+          )}`,
+        );
+        await this.options.events.subscribe({
+          topics,
+          id: `catalog-backend-module-incremental-ingestion:${provider.getProviderName()}`,
+          onEvent: evt => engine.onEvent(evt),
+        });
+      }
     } catch (error) {
       logger.warn(
         `Failed to initialize incremental ingestion provider ${provider.getProviderName()}, ${stringifyError(

plugins/catalog-backend-module-incremental-ingestion/src/module/catalogModuleIncrementalIngestionEntityProvider.ts

@@ -25,6 +25,7 @@ import {
   IncrementalEntityProviderOptions,
 } from '@backstage/plugin-catalog-backend-module-incremental-ingestion';
 import { WrapperProviders } from './WrapperProviders';
+import { eventsServiceRef } from '@backstage/plugin-events-node';
 
 /**
  * @public
@@ -106,6 +107,7 @@ export const catalogModuleIncrementalIngestionEntityProvider =
           httpRouter: coreServices.httpRouter,
           logger: coreServices.logger,
           scheduler: coreServices.scheduler,
+          events: eventsServiceRef,
         },
         async init({
           catalog,
@@ -114,6 +116,7 @@ export const catalogModuleIncrementalIngestionEntityProvider =
           httpRouter,
           logger,
           scheduler,
+          events,
         }) {
           const client = await database.getClient();
 
@@ -122,6 +125,7 @@ export const catalogModuleIncrementalIngestionEntityProvider =
             logger,
             client,
             scheduler,
+            events,
           });
 
           for (const entry of addedProviders) {