diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts b/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts index 6d2b3e2be5..aa2d4fe6d2 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts @@ -26,6 +26,7 @@ import { applyDatabaseMigrations } from '../database/migrations'; import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager'; import { IncrementalProviderRouter } from '../router/routes'; import { Deferred } from '../util'; +import { EventParams, EventSubscriber } from '@backstage/plugin-events-node'; /** @public */ export class IncrementalCatalogBuilder { @@ -71,13 +72,15 @@ export class IncrementalCatalogBuilder { addIncrementalEntityProvider( provider: IncrementalEntityProvider, options: IncrementalEntityProviderOptions, - ) { + ): EventSubscriber { const { burstInterval, burstLength, restLength } = options; const { logger: catalogLogger, scheduler } = this.env; const ready = this.ready; const manager = this.manager; + let engine: IncrementalIngestionEngine; + this.builder.addEntityProvider({ getProviderName: provider.getProviderName.bind(provider), async connect(connection) { @@ -87,7 +90,7 @@ export class IncrementalCatalogBuilder { logger.info(`Connecting`); - const engine = new IncrementalIngestionEngine({ + engine = new IncrementalIngestionEngine({ ...options, ready, manager, @@ -112,5 +115,12 @@ export class IncrementalCatalogBuilder { }); }, }); + + return { + onEvent: (params: EventParams) => engine.onEvent(params), + get supportsEventTopics() { + return engine.supportsEventTopics; + }, + }; } }