core io tests #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open

Mcnoble1 wants to merge 23 commits into risevest:main from Mcnoble1:festus

package.json

-Original file line number
+Diff line change
@@ Expand Up / @@ -37,7 +37,6 @@ @@
         "ts-node": "^10.9.1",
         "typescript": "^4.9.4"
       },
-      "dependencies": {},
       "engines": {
         "node": ">=18.0.0"
       }
@@ Expand Down @@

src/io.ts

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -2,7 +2,7 @@ import { MemoryStorage, Storage } from "./storage";
  
    import { Worker } from "./worker";

    export class BufIO<T, U> {

    export class IO<T, U> {

      private storage: Storage<T>;

      private worker: Worker<T, U>;

      private flushInterval: number;

    @@ -17,22 +17,35 @@ export class BufIO<T, U> {
  
        flushInterval?: number;

        onError?: (err: Error, records: T[]) => void;

      }) {

        if (!config.worker) {

          throw new Error("Worker must be provided");

        }

        if (config.batchSize <= 0) {

          throw new Error("batchSize must be greater than 0");

        }

        this.storage = config.storage ?? new MemoryStorage<T>();

        this.worker = config.worker;

        this.batchSize = config.batchSize;

        this.flushInterval = config.flushInterval ?? 5000;

        this.onError = config.onError;

      }

      public push(record: T) {

      public async push(record: T) {

        if (record === null || record === undefined) return;

        this.storage.put(record);

      }

      public start() {

        this.intervalId = setInterval(() => this.flush(), this.flushInterval);

      public async start() {

        if (this.intervalId && !(this.intervalId as any)._destroyed === false)

          return;

        if (!this.intervalId || (this.intervalId as any)._destroyed) {

          this.intervalId = setInterval(() => {

            this.flush();

          }, this.flushInterval);

        }

      }

      public stop() {

      public async stop() {

        if (this.intervalId) clearInterval(this.intervalId);

      }

0 src/worker.ts → src/worker/base.ts

File renamed without changes.

src/worker/index.ts

Original file line number	Diff line number	Diff line change
		@@ -0,0 +1 @@
		export * from "./base";

tests/helpers/worker.ts

-Original file line number
+Diff line change
@@ -0,0 +1,10 @@
+    import { Worker } from "../../src/worker/base";
+    export class TestWorker<T> implements Worker<T, T> {
+      public processed: T[] = [];
+      async work(records: T[]): Promise<T[]> {
+        this.processed.push(...records);
+        return records;
+      }
+    }

tests/integration/io.spec.ts

-Original file line number
+Diff line change
@@ -0,0 +1,130 @@
+    import chai, { expect } from "chai";
+    import chaiAsPromised from "chai-as-promised";
+    import { IO } from "../../src/io";
+    import { MemoryStorage } from "../../src/storage/memory";
+    import { Worker } from "../../src/worker";
+    import { TestWorker } from "../helpers/worker";
+    chai.use(chaiAsPromised);
+    type RecordType = { id: number; value: string };
+    describe("IO Integration", () => {
+      let storage: MemoryStorage<RecordType>;
+      let worker: TestWorker<RecordType>;
+      let io: IO<RecordType, RecordType>;
+      beforeEach(() => {
+        storage = new MemoryStorage<RecordType>();
+        worker = new TestWorker();
+        io = new IO({
+          storage,
+          worker,
+          batchSize: 2,
+          flushInterval: 100,
+        });
+      });
+      afterEach(() => {
+        io.stop();
+      });
+      it("should initialize with MemoryStorage and Worker", () => {
+        expect(io).to.have.property("storage").that.equals(storage);
+        expect(io).to.have.property("worker").that.equals(worker);
+      });
+      it("should push records into storage correctly", () => {
+        io.push({ id: 1, value: "A" });
+        io.push({ id: 2, value: "B" });
+        const records = storage.get(10);
+        expect(records).to.deep.equal([
+          { id: 1, value: "A" },
+          { id: 2, value: "B" },
+        ]);
+      });
+      it("should flush records manually", async () => {
+        io.push({ id: 1, value: "A" });
+        io.push({ id: 2, value: "B" });
+        await (io as any).flush();
+        expect(worker.processed).to.deep.equal([
+          { id: 1, value: "A" },
+          { id: 2, value: "B" },
+        ]);
+        const remaining = storage.get(10);
+        expect(remaining).to.be.empty;
+      });
+      it("should automatically flush records at interval", async () => {
+        io.start();
+        io.push({ id: 1, value: "X" });
+        io.push({ id: 2, value: "Y" });
+        await new Promise((resolve) => setTimeout(resolve, 200)); // wait for interval flush
+        expect(worker.processed).to.deep.include.members([
+          { id: 1, value: "X" },
+          { id: 2, value: "Y" },
+        ]);
+      });
+      it("should respect batchSize when flushing", async () => {
+        io.push({ id: 1, value: "one" });
+        io.push({ id: 2, value: "two" });
+        io.push({ id: 3, value: "three" });
+        await (io as any).flush();
+        // first 2 processed, 1 left in storage
+        expect(worker.processed).to.deep.equal([
+          { id: 1, value: "one" },
+          { id: 2, value: "two" },
+        ]);
+        expect(storage.get(10)).to.deep.equal([{ id: 3, value: "three" }]);
+      });
+      it("should handle worker errors gracefully", async () => {
+        class FailingWorker implements Worker<RecordType, RecordType> {
+          async work(): Promise<RecordType[]> {
+            throw new Error("worker failed");
+          }
+        }
+        const failingIO = new IO({
+          storage: new MemoryStorage<RecordType>(),
+          worker: new FailingWorker(),
+          batchSize: 2,
+          flushInterval: 100,
+        });
+        failingIO.push({ id: 99, value: "bad" });
+        await expect((failingIO as any).flush()).to.be.fulfilled;
+      });
+      it("should stop flushing when stop() is called", async () => {
+        io.start();
+        io.stop();
+        io.push({ id: 1, value: "stop-test" });
+        await new Promise((resolve) => setTimeout(resolve, 200));
+        expect(worker.processed).to.not.deep.include({ id: 1, value: "stop-test" });
+      });
+      it("should handle concurrent push and flush safely", async () => {
+        io.push({ id: 1, value: "X" });
+        const flushPromise = (io as any).flush();
+        io.push({ id: 2, value: "Y" });
+        await flushPromise;
+        // at least one record should remain in storage, but none should be lost
+        const all = [...worker.processed, ...storage.get(10)];
+        expect(all).to.deep.include.members([
+          { id: 1, value: "X" },
+          { id: 2, value: "Y" },
+        ]);
+      });
+    });

tests/io.spec.ts

This file was deleted.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

core io tests #5

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

core io tests #5

Are you sure you want to change the base?

Uh oh!

core io tests #5

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!