moved around

Signed-off-by: Fredrik Adelöw <freben@gmail.com>

plugins/catalog-backend/src/processing/TaskPipeline.test.ts

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-import { startTaskPipeline } from './TaskPipeline';
+import { startTaskPipeline, createBarrier } from './TaskPipeline';
 
 function createLimitedLoader(count: number, loadDelay?: number) {
   const items = new Array(count).fill(0).map((_, index) => index);
@@ -119,3 +119,88 @@ describe('startTaskPipeline', () => {
     }).toThrow('must be lower');
   });
 });
+
+describe('createBarrier', () => {
+  const tick = (millis: number) =>
+    new Promise(resolve => setTimeout(resolve, millis));
+
+  it('abandons a wait after the timeout expires', async () => {
+    const abortController = new AbortController();
+    const signal = abortController.signal;
+    const barrier = createBarrier({ waitTimeoutMillis: 100, signal });
+
+    const fn1 = jest.fn();
+    barrier.wait().then(fn1);
+
+    await tick(0);
+    expect(fn1).not.toHaveBeenCalled();
+
+    await tick(50);
+    expect(fn1).not.toHaveBeenCalled();
+
+    // start a new wait mid-way through the timeout
+    // should NOT resolve when the first one times out
+    const fn2 = jest.fn();
+    barrier.wait().then(fn2);
+
+    await tick(0);
+    expect(fn2).not.toHaveBeenCalled();
+
+    await tick(50);
+    expect(fn1).toHaveBeenCalledTimes(1);
+    expect(fn2).not.toHaveBeenCalled();
+
+    await tick(50);
+    expect(fn1).toHaveBeenCalledTimes(1);
+    expect(fn2).toHaveBeenCalledTimes(1);
+  });
+
+  it('abandons a wait after aborted', async () => {
+    const abortController = new AbortController();
+    const signal = abortController.signal;
+    const barrier = createBarrier({ waitTimeoutMillis: 100, signal });
+
+    const fn1 = jest.fn();
+    barrier.wait().then(fn1);
+
+    // should resolve immediately, not after timeout
+    await tick(0);
+    expect(fn1).not.toHaveBeenCalled();
+    abortController.abort();
+    await tick(0);
+    expect(fn1).toHaveBeenCalledTimes(1);
+
+    // subsequent waits should be immediate no matter what
+    const fn2 = jest.fn();
+    barrier.wait().then(fn2);
+    await tick(0);
+    expect(fn2).toHaveBeenCalledTimes(1);
+  });
+
+  it('release immediately unblocks all waits', async () => {
+    const abortController = new AbortController();
+    const signal = abortController.signal;
+    const barrier = createBarrier({ waitTimeoutMillis: 100, signal });
+
+    const fn1 = jest.fn();
+    barrier.wait().then(fn1);
+
+    await tick(50);
+    expect(fn1).not.toHaveBeenCalled();
+
+    // start a new wait mid-way through the timeout
+    // SHOULD resolve when releasing
+    const fn2 = jest.fn();
+    barrier.wait().then(fn2);
+
+    await tick(0);
+    expect(fn1).not.toHaveBeenCalled();
+    expect(fn2).not.toHaveBeenCalled();
+
+    barrier.release();
+
+    await tick(0);
+    expect(fn1).toHaveBeenCalledTimes(1);
+    expect(fn2).toHaveBeenCalledTimes(1);
+  });
+});

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

@@ -14,8 +14,6 @@
  * limitations under the License.
  */
 
-import { createBarrier } from './util';
-
 const DEFAULT_POLLING_INTERVAL_MS = 1000;
 
 type Options<T> = {
@@ -80,12 +78,12 @@ export function startTaskPipeline<T>(options: Options<T>) {
   const abortController = new AbortController();
   const abortSignal = abortController.signal;
 
-  async function pipelineLoop() {
-    const barrier = createBarrier({
-      waitTimeoutMillis: pollingIntervalMs,
-      signal: abortSignal,
-    });
+  const barrier = createBarrier({
+    waitTimeoutMillis: pollingIntervalMs,
+    signal: abortSignal,
+  });
 
+  async function pipelineLoop() {
     while (!abortSignal.aborted) {
       if (state.inFlightCount <= lowWatermark) {
         const loadCount = highWatermark - state.inFlightCount;
@@ -127,5 +125,56 @@ export function startTaskPipeline<T>(options: Options<T>) {
 
   return () => {
     abortController.abort();
+    barrier.destroy();
+  };
+}
+
+/**
+ * Creates a barrier with a timeout, that can be awaited or prematurely
+ * released either manually or by an abort signal.
+ */
+export function createBarrier(options: {
+  waitTimeoutMillis: number;
+  signal: AbortSignal;
+}): {
+  wait: () => Promise<void>;
+  release: () => void;
+  destroy: () => void;
+} {
+  const { waitTimeoutMillis, signal } = options;
+  const resolvers = new Set<() => void>();
+
+  function wait() {
+    if (signal.aborted || !(waitTimeoutMillis > 0)) {
+      return Promise.resolve();
+    }
+
+    return new Promise<void>(resolve => {
+      const timeoutHandle = setTimeout(done, waitTimeoutMillis);
+
+      function done() {
+        resolvers.delete(done);
+        clearTimeout(timeoutHandle);
+        resolve();
+      }
+
+      resolvers.add(done);
+    });
+  }
+
+  function release() {
+    const resolversToCall = new Set(resolvers);
+    resolvers.clear();
+    for (const resolver of resolversToCall) {
+      resolver();
+    }
+  }
+
+  signal.addEventListener('abort', release);
+
+  return {
+    wait,
+    release,
+    destroy: () => signal.removeEventListener('abort', release),
   };
 }

plugins/catalog-backend/src/processing/util.test.ts

@@ -1,102 +0,0 @@
-/*
- * 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 { createBarrier } from './util';
-
-describe('createBarrier', () => {
-  const tick = (millis: number) =>
-    new Promise(resolve => setTimeout(resolve, millis));
-
-  it('abandons a wait after the timeout expires', async () => {
-    const abortController = new AbortController();
-    const signal = abortController.signal;
-    const barrier = createBarrier({ waitTimeoutMillis: 100, signal });
-
-    const fn1 = jest.fn();
-    barrier.wait().then(fn1);
-
-    await tick(0);
-    expect(fn1).not.toHaveBeenCalled();
-
-    await tick(50);
-    expect(fn1).not.toHaveBeenCalled();
-
-    // start a new wait mid-way through the timeout
-    // should NOT resolve when the first one times out
-    const fn2 = jest.fn();
-    barrier.wait().then(fn2);
-
-    await tick(0);
-    expect(fn2).not.toHaveBeenCalled();
-
-    await tick(50);
-    expect(fn1).toHaveBeenCalledTimes(1);
-    expect(fn2).not.toHaveBeenCalled();
-
-    await tick(50);
-    expect(fn1).toHaveBeenCalledTimes(1);
-    expect(fn2).toHaveBeenCalledTimes(1);
-  });
-
-  it('abandons a wait after aborted', async () => {
-    const abortController = new AbortController();
-    const signal = abortController.signal;
-    const barrier = createBarrier({ waitTimeoutMillis: 100, signal });
-
-    const fn1 = jest.fn();
-    barrier.wait().then(fn1);
-
-    // should resolve immediately, not after timeout
-    await tick(0);
-    expect(fn1).not.toHaveBeenCalled();
-    abortController.abort();
-    await tick(0);
-    expect(fn1).toHaveBeenCalledTimes(1);
-
-    // subsequent waits should be immediate no matter what
-    const fn2 = jest.fn();
-    barrier.wait().then(fn2);
-    await tick(0);
-    expect(fn2).toHaveBeenCalledTimes(1);
-  });
-
-  it('release immediately unblocks all waits', async () => {
-    const abortController = new AbortController();
-    const signal = abortController.signal;
-    const barrier = createBarrier({ waitTimeoutMillis: 100, signal });
-
-    const fn1 = jest.fn();
-    barrier.wait().then(fn1);
-
-    await tick(50);
-    expect(fn1).not.toHaveBeenCalled();
-
-    // start a new wait mid-way through the timeout
-    // SHOULD resolve when releasing
-    const fn2 = jest.fn();
-    barrier.wait().then(fn2);
-
-    await tick(0);
-    expect(fn1).not.toHaveBeenCalled();
-    expect(fn2).not.toHaveBeenCalled();
-
-    barrier.release();
-
-    await tick(0);
-    expect(fn1).toHaveBeenCalledTimes(1);
-    expect(fn2).toHaveBeenCalledTimes(1);
-  });
-});

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

@@ -86,50 +86,3 @@ export function isObject(value: JsonValue | undefined): value is JsonObject {
 export const validateEntity = entitySchemaValidator();
 
 export const validateEntityEnvelope = entityEnvelopeSchemaValidator();
-
-/**
- * Creates a barrier with a timeout, that can be awaited or prematurely
- * released either manually or by an abort signal.
- */
-export function createBarrier(options: {
-  waitTimeoutMillis: number;
-  signal: AbortSignal;
-}): {
-  wait: () => Promise<void>;
-  release: () => void;
-} {
-  const { waitTimeoutMillis, signal } = options;
-  const resolvers: Array<() => void> = [];
-
-  function wait() {
-    if (signal.aborted || !(waitTimeoutMillis > 0)) {
-      return Promise.resolve();
-    }
-
-    return new Promise<void>(resolve => {
-      const timeoutHandle = setTimeout(done, waitTimeoutMillis);
-
-      function done() {
-        const index = resolvers.indexOf(done);
-        if (index !== -1) {
-          resolvers.splice(index, 1);
-        }
-        clearTimeout(timeoutHandle);
-        resolve();
-      }
-
-      resolvers.push(done);
-    });
-  }
-
-  function release() {
-    const resolversToCall = resolvers.splice(0, resolvers.length);
-    for (const resolver of resolversToCall) {
-      resolver();
-    }
-  }
-
-  signal.addEventListener('abort', release);
-
-  return { wait, release };
-}