Added Closeable collection and shutdown mechanism

Author: rpanic

packages/sequencer/src/index.ts

@@ -6,6 +6,7 @@ export * from "./mempool/private/PrivateMempool";
 export * from "./sequencer/executor/Sequencer";
 export * from "./sequencer/executor/Sequenceable";
 export * from "./sequencer/builder/SequencerModule";
+export * from "./sequencer/builder/Closeable";
 export * from "./worker/flow/Flow";
 export * from "./worker/flow/Task";
 export * from "./worker/flow/JSONTaskSerializer";

packages/sequencer/src/protocol/production/trigger/TimedBlockTrigger.ts

@@ -2,7 +2,7 @@ import { inject, injectable } from "tsyringe";
 import { injectOptional, log } from "@proto-kit/common";
 import gcd from "compute-gcd";
 
-import { Closeable } from "../../../worker/queue/TaskQueue";
+import { Closeable } from "../../../sequencer/builder/Closeable";
 import { BatchProducerModule } from "../BatchProducerModule";
 import { Mempool } from "../../../mempool/Mempool";
 import { BlockQueue } from "../../../storage/repositories/BlockStorage";

packages/sequencer/src/sequencer/builder/Closeable.ts

@@ -0,0 +1,9 @@
+import { implement } from "@proto-kit/common/dist/config/injectAlias";
+
+export interface Closeable {
+  close: () => Promise<void>;
+}
+
+export function closeable() {
+  return implement<Closeable>("Closeable");
+}

packages/sequencer/src/sequencer/executor/Sequencer.ts

@@ -18,6 +18,7 @@ import {
 import { DependencyContainer, injectable } from "tsyringe";
 
 import { SequencerModule } from "../builder/SequencerModule";
+import { Closeable } from "../builder/Closeable";
 
 import { Sequenceable } from "./Sequenceable";
 
@@ -86,4 +87,13 @@ export class Sequencer<Modules extends SequencerModulesRecord>
       await sequencerModule.start();
     }
   }
+
+  public async close() {
+    log.info("Closing sequencer...");
+    const closeables = this.container.resolveAll<Closeable>("Closeable");
+    await Promise.all(
+      closeables.map(async (closeable) => await closeable.close())
+    );
+    log.info("Sequencer closed");
+  }
 }

packages/sequencer/src/worker/flow/Flow.ts

@@ -1,7 +1,8 @@
 import { inject, injectable } from "tsyringe";
 import { log, mapSequential } from "@proto-kit/common";
 
-import { Closeable, InstantiatedQueue, TaskQueue } from "../queue/TaskQueue";
+import { InstantiatedQueue, TaskQueue } from "../queue/TaskQueue";
+import { Closeable } from "../../sequencer/builder/Closeable";
 
 import { Task, TaskPayload } from "./Task";
 

packages/sequencer/src/worker/queue/LocalTaskQueue.ts

@@ -2,8 +2,9 @@ import { log, mapSequential, noop } from "@proto-kit/common";
 
 import { sequencerModule } from "../../sequencer/builder/SequencerModule";
 import { TaskPayload } from "../flow/Task";
+import { Closeable } from "../../sequencer/builder/Closeable";
 
-import { Closeable, InstantiatedQueue, TaskQueue } from "./TaskQueue";
+import { InstantiatedQueue, TaskQueue } from "./TaskQueue";
 import { ListenerList } from "./ListenerList";
 import { AbstractTaskQueue } from "./AbstractTaskQueue";
 

packages/sequencer/src/worker/queue/TaskQueue.ts

@@ -1,4 +1,5 @@
 import { TaskPayload } from "../flow/Task";
+import { Closeable } from "../../sequencer/builder/Closeable";
 
 /**
  * Definition of a connection-object that can generate queues and workers
@@ -13,11 +14,6 @@ export interface TaskQueue {
     options?: { concurrency?: number }
   ) => Closeable;
 }
-
-export interface Closeable {
-  close: () => Promise<void>;
-}
-
 /**
  * Object that abstracts a concrete connection to a queue instance.
  */

packages/sequencer/src/worker/worker/FlowTaskWorker.ts

@@ -1,9 +1,10 @@
 import { log } from "@proto-kit/common";
 
-import { Closeable, TaskQueue } from "../queue/TaskQueue";
+import { TaskQueue } from "../queue/TaskQueue";
 import { Task, TaskPayload } from "../flow/Task";
 import { AbstractStartupTask } from "../flow/AbstractStartupTask";
 import { UnpreparingTask } from "../flow/UnpreparingTask";
+import { Closeable } from "../../sequencer/builder/Closeable";
 
 const errors = {
   notComputable: (name: string) =>

packages/sequencer/src/worker/worker/startup/WorkerRegistrationFlow.ts

@@ -1,7 +1,7 @@
 import { injectable } from "tsyringe";
 import { log } from "@proto-kit/common";
 
-import { Closeable } from "../../queue/TaskQueue";
+import { Closeable } from "../../../sequencer/builder/Closeable";
 import { FlowCreator } from "../../flow/Flow";
 
 import {

packages/sequencer/test/sequencer/executor/Sequencer.test.ts

@@ -0,0 +1,69 @@
+import "reflect-metadata";
+import { DependencyFactory, noop, sleep } from "@proto-kit/common";
+import { jest } from "@jest/globals";
+import { container } from "tsyringe";
+
+import {
+  Closeable,
+  closeable,
+  Sequencer,
+  sequencerModule,
+  SequencerModule,
+} from "../../../src";
+
+describe("Sequencer close", () => {
+  it("should close all services", async () => {
+    const spyFn = jest.fn<() => void>();
+
+    @closeable()
+    @sequencerModule()
+    class CloseableModule extends SequencerModule implements Closeable {
+      public async start(): Promise<void> {
+        noop();
+      }
+
+      public async close() {
+        await sleep(200);
+        spyFn.call(undefined);
+      }
+    }
+
+    @sequencerModule()
+    class DependencyFactoryModule
+      extends SequencerModule
+      implements DependencyFactory
+    {
+      public async start(): Promise<void> {
+        noop();
+      }
+
+      dependencies() {
+        return {
+          Dep2: {
+            useClass: CloseableModule,
+          },
+        };
+      }
+    }
+
+    const sequencer = new (Sequencer.from({
+      modules: {
+        Foo: CloseableModule,
+        Bar: CloseableModule,
+        D: DependencyFactoryModule,
+      },
+    }))();
+    sequencer.create(() => container.createChildContainer());
+    sequencer.configure({
+      Foo: {},
+      Bar: {},
+      D: {},
+    });
+
+    sequencer.resolve("D");
+
+    await sequencer.close();
+
+    expect(spyFn).toHaveBeenCalledTimes(3);
+  });
+});