catalog-backend: refactor entityProviders connect

Signed-off-by: Johan Haals <johan.haals@gmail.com>

plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts

@@ -14,11 +14,7 @@
  * limitations under the License.
  */
 
-import {
-  Entity,
-  entityEnvelopeSchemaValidator,
-  stringifyEntityRef,
-} from '@backstage/catalog-model';
+import { stringifyEntityRef } from '@backstage/catalog-model';
 import { assertError, serializeError } from '@backstage/errors';
 import { Hash } from 'crypto';
 import stableStringify from 'fast-json-stable-stringify';
@@ -32,68 +28,15 @@ import {
 } from '../processing/types';
 import { Stitcher } from '../stitching/Stitcher';
 import { startTaskPipeline } from './TaskPipeline';
-import {
-  EntityProvider,
-  EntityProviderConnection,
-  EntityProviderMutation,
-} from '../providers/types';
 
 const CACHE_TTL = 5;
 
-class Connection implements EntityProviderConnection {
-  readonly validateEntityEnvelope = entityEnvelopeSchemaValidator();
-
-  constructor(
-    private readonly config: {
-      id: string;
-      processingDatabase: ProcessingDatabase;
-    },
-  ) {}
-
-  async applyMutation(mutation: EntityProviderMutation): Promise<void> {
-    const db = this.config.processingDatabase;
-
-    if (mutation.type === 'full') {
-      this.check(mutation.entities.map(e => e.entity));
-      await db.transaction(async tx => {
-        await db.replaceUnprocessedEntities(tx, {
-          sourceKey: this.config.id,
-          type: 'full',
-          items: mutation.entities,
-        });
-      });
-    } else if (mutation.type === 'delta') {
-      this.check(mutation.added.map(e => e.entity));
-      this.check(mutation.removed.map(e => e.entity));
-      await db.transaction(async tx => {
-        await db.replaceUnprocessedEntities(tx, {
-          sourceKey: this.config.id,
-          type: 'delta',
-          added: mutation.added,
-          removed: mutation.removed,
-        });
-      });
-    }
-  }
-
-  private check(entities: Entity[]) {
-    for (const entity of entities) {
-      try {
-        this.validateEntityEnvelope(entity);
-      } catch (e) {
-        throw new TypeError(`Malformed entity envelope, ${e}`);
-      }
-    }
-  }
-}
-
 export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
   private readonly tracker = progressTracker();
   private stopFunc?: () => void;
 
   constructor(
     private readonly logger: Logger,
-    private readonly entityProviders: EntityProvider[],
     private readonly processingDatabase: ProcessingDatabase,
     private readonly orchestrator: CatalogProcessingOrchestrator,
     private readonly stitcher: Stitcher,
@@ -106,15 +49,6 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
       throw new Error('Processing engine is already started');
     }
 
-    for (const provider of this.entityProviders) {
-      await provider.connect(
-        new Connection({
-          id: provider.getProviderName(),
-          processingDatabase: this.processingDatabase,
-        }),
-      );
-    }
-
     this.stopFunc = startTaskPipeline<RefreshStateItem>({
       lowWatermark: 5,
       highWatermark: 10,

plugins/catalog-backend/src/processing/connectEntityProviders.ts

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2021 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 {
+  Entity,
+  entityEnvelopeSchemaValidator,
+} from '@backstage/catalog-model';
+import { ProcessingDatabase } from '../database/types';
+import {
+  EntityProvider,
+  EntityProviderConnection,
+  EntityProviderMutation,
+} from '../providers/types';
+
+class Connection implements EntityProviderConnection {
+  readonly validateEntityEnvelope = entityEnvelopeSchemaValidator();
+
+  constructor(
+    private readonly config: {
+      id: string;
+      processingDatabase: ProcessingDatabase;
+    },
+  ) {}
+
+  async applyMutation(mutation: EntityProviderMutation): Promise<void> {
+    const db = this.config.processingDatabase;
+
+    if (mutation.type === 'full') {
+      this.check(mutation.entities.map(e => e.entity));
+      await db.transaction(async tx => {
+        await db.replaceUnprocessedEntities(tx, {
+          sourceKey: this.config.id,
+          type: 'full',
+          items: mutation.entities,
+        });
+      });
+    } else if (mutation.type === 'delta') {
+      this.check(mutation.added.map(e => e.entity));
+      this.check(mutation.removed.map(e => e.entity));
+      await db.transaction(async tx => {
+        await db.replaceUnprocessedEntities(tx, {
+          sourceKey: this.config.id,
+          type: 'delta',
+          added: mutation.added,
+          removed: mutation.removed,
+        });
+      });
+    }
+  }
+
+  private check(entities: Entity[]) {
+    for (const entity of entities) {
+      try {
+        this.validateEntityEnvelope(entity);
+      } catch (e) {
+        throw new TypeError(`Malformed entity envelope, ${e}`);
+      }
+    }
+  }
+}
+
+export async function connectEntityProviders(
+  db: ProcessingDatabase,
+  providers: EntityProvider[],
+) {
+  await Promise.all(
+    providers.map(async provider => {
+      const connection = new Connection({
+        id: provider.getProviderName(),
+        processingDatabase: db,
+      });
+      return provider.connect(connection);
+    }),
+  );
+}

plugins/catalog-backend/src/service/NextCatalogBuilder.ts

@@ -80,6 +80,7 @@ import { DefaultCatalogRulesEnforcer } from '../ingestion/CatalogRules';
 import { Config } from '@backstage/config';
 import { Logger } from 'winston';
 import { LocationService } from './types';
+import { connectEntityProviders } from '../processing/connectEntityProviders';
 
 export type CatalogEnvironment = {
   logger: Logger;
@@ -364,7 +365,6 @@ export class NextCatalogBuilder {
 
     const processingEngine = new DefaultCatalogProcessingEngine(
       logger,
-      entityProviders,
       processingDatabase,
       orchestrator,
       stitcher,
@@ -390,6 +390,8 @@ export class NextCatalogBuilder {
       config,
     });
 
+    await connectEntityProviders(processingDatabase, entityProviders);
+
     return {
       entitiesCatalog,
       locationsCatalog,