TypeScript Observers for codegen

CHANGELOG.md

@@ -15,6 +15,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - `getOperationsToImplementOrDeclareFromInterfacesWithoutParents`.
   - `getOperationsToImplementOrDeclare`.
 
+TypeScript:
+- Method Observers
+  - `connector.subscribe('export', observer: Observer<string>)`
+  - `connector.subscribe('import', observer: Observer<DatasetExt[]>)`
+- Released versions 1.0.0-alpha.10, 1.0.0-alpha-11
+
 ### Changed
 TypeScript:
 - Ensure the entry point, `src/index.ts`, is always regenerated on fresh builds.

src/org/datafoodconsortium/connector/codegen/typescript/static/CHANGELOG.md

@@ -12,6 +12,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - None
 
+## [1.0.0-alpha.11] - 2025-05-01
+
+### Added
+- Method Observers
+  - `connector.subscribe('export', observer: Observer<string>)`
+  - `connector.subscribe('import', observer: Observer<DatasetExt[]>)`
+
 ## [1.0.0-alpha.10] - 2025-04-28
 
 ### Changed

src/org/datafoodconsortium/connector/codegen/typescript/static/README.md

@@ -292,6 +292,110 @@ const order: string = "http://www.datafoodconsortium.org/ontologies/DFC_Business
 const IOrder | undefined = await connector.importOneTyped<IOrder>(jsonAsAString, { only: order });
 ```
 
+## Method Observers
+Certain methods permit observers to listen events and perform side effects
+whenever those methods are called. For instance, you can subscribe to all
+`"export"` events by registering an object with three methods: a `.next()`
+method, which fires each time `connector.export()` is invoked and receives
+as its single argument the same JSON payload returned by the export method; an
+optional `.error()` method, in case the export fails; and an optional
+`.complete()` method, which is called without any arguments whenever the
+observer unsubscribes or in the case that the Connector cancels it for some
+reason.
+
+```js
+const exportedJson = new Map();
+const exportObserver = {
+  next(json) {
+    exportedJson.set(Date.now(), json);
+  },
+  error(e) { console.error(e.message); },
+  complete() {
+    console.log('Total exports:', exportedJson.size);
+  },
+}
+const subscription = connector.subscribe('export', exportObserver);
+
+connector.export(someDataSet); // exportedJson.size => 1
+connector.export(anotherDataSet); // exportedJson.size => 2
+subscription.unsubscribe(); // CONSOLE: 'Total exports: 2'
+```
+
+The following string values can be passed as the first argument to
+`Connector.subscribe()`, supporting the corresponding methods/events:
+
+- `"export"`
+- `"import"`
+
+An observer can also implement the `Observer<T>` interface as a class for more
+nuanced, stateful behavior, and so thatmultiple instantiations can each
+subscribe and unsubscribe to events separately:
+
+```ts
+class ImportObserver<T extends DatasetExt[]> implements Observer<T> {
+  type: string;
+  baseUrl: URL;
+  noisy = false;
+  #imported: QuadExt[] = [];
+
+  constructor(semType: string|Semanticable, baseUrl: string, noisy?: boolean) {
+    if (typeof semType === 'string') this.type = semType;
+    else this.type = semType.getSemanticType();
+    this.baseUrl = new URL(baseUrl);
+    if (typeof noisy === "boolean") this.noisy = noisy;
+  }
+
+  next(datasets: DatasetExt[]) {
+    const count = datasets.reduce((i, dSet) => {
+      return i + dSet.reduce((j, quad) => {
+        if (quad.subject.termType !== this.type) return j;
+        this.#imported.push(quad);
+        return j + 1;
+      }, 0);
+    }, 0);
+    if (this.noisy && count > 0) {
+      console.log(`Imported ${this.type} ${count}x.`);
+    }
+  }
+
+  error(e: Error) {
+    if (this.noisy) console.error('Whoops!', e.message);
+  }
+
+  complete() {
+    this.#imported.forEach((quad) => {
+      const opts = { method: 'POST', body: JSON.stringify(quad)};
+      fetch(`${this.baseUrl}/${this.type}`, opts);
+    });
+    if (this.noisy) {
+      const total = this.#imported.length;
+      const msg = this.noisy && total > 0
+        ? `Imported ${this.type} a total of ${total} times! 🙌`
+        : `Didn't import a single ${this.type}! 😭`;
+      console.log(msg);
+    }
+  }
+}
+```
+
+Multiple observers can also subscribe to the same method/event simultaneously to
+to control their behaviors and attributes separately:
+
+```js
+// Use the ImportObserver above to watch for different data types.
+const api = 'https://api.example.net/v1/';
+const addrObs = new ImportObserver('dfc-b:Address', api, true);
+const catalogObs = new ImportObserver('"dfc-b:CatalogItem"', api);
+const catalogSub = connector.subscribe('import', catalogObs);
+const addrSubAPIv1 = connector.subscribe('import', addrObs);
+
+// Add new observers with different behaviors or modify others.
+const api2 = 'https://api.example.net/v2/';
+const addrObsAPIv2 = new ImportObserver('dfc-b:Address', api2);
+const addrSubAPIv2 = connector.subscribe('import', addrObsAPIv2);
+addrObsAPIv1.noisy = false;
+```
+
 ## Configure
 
 You can adapt different components of the connector to your needs with the following connector methods:

src/org/datafoodconsortium/connector/codegen/typescript/static/package.json

@@ -4,10 +4,10 @@
   "author": "Maxime Lecoq",
   "license": "MIT",
   "type": "module",
-  "version": "1.0.0-alpha.10",
+  "version": "1.0.0-alpha.11",
   "repository": {
     "type": "git",
-    "url": "https://github.com/datafoodconsortium/connector-typescript.git",
+    "url": "git+https://github.com/datafoodconsortium/connector-typescript.git",
     "directory": "/"
   },
   "scripts": {

src/org/datafoodconsortium/connector/codegen/typescript/static/src/Connector.ts

@@ -16,6 +16,7 @@ import IConnectorImporter from "./IConnectorImporter";
 import IConnectorImportOptions from "./IConnectorImportOptions.js";
 import IConnectorStore from "./IConnectorStore";
 import IGetterOptions from "./IGetterOptions.js";
+import type { Observer } from "./observer.js";
 
 // Generated Code
 import IAddress from "./IAddress.js";
@@ -41,12 +42,24 @@ import IPlannedConsumptionFlow from "./IPlannedConsumptionFlow.js";
 import IPlannedProductionFlow from "./IPlannedProductionFlow.js";
 import IDefinedProduct from "./IDefinedProduct.js";
 
+// The keys map to the name of the method to the private member where the
+// corresponding observable object is actually stored.
+const ConnectorObservables = {
+    export: "exporter",
+    import: "importer",
+} as const;
+type ConnectorObservables = typeof ConnectorObservables;
+type ConnectorObservableKeys = keyof ConnectorObservables;
+type ConnectorObservableMethods = ConnectorObservables[ConnectorObservableKeys];
+type ConnectorObservableStrings = ConnectorObservableKeys | ConnectorObservableMethods;
+
 export default class Connector implements IConnector {
 
     public FACETS?: ISKOSConcept;
     public MEASURES?: ISKOSConcept;
     public PRODUCT_TYPES?: ISKOSConcept;
     public VOCABULARY?: ISKOSConcept;
+    public OBSERVABLES = ConnectorObservables;
 
     private semantizer: ISemantizer;
     private fetchFunction: (semanticId: string) => Promise<Response>;
@@ -195,6 +208,13 @@ export default class Connector implements IConnector {
         return this.factory;
     }
 
+    public subscribe(event: ConnectorObservableStrings, observer: Observer<any>) {
+        const observable = event in this.OBSERVABLES
+            ? this.OBSERVABLES[event as ConnectorObservableKeys]
+            : event as ConnectorObservableMethods;
+        return this[observable].subscribe(observer);
+    }
+
     public async import(data: string, options?: IConnectorImportOptions): Promise<Array<Semanticable>> {
         return new Promise(async (resolve, reject) => {
             try { 

src/org/datafoodconsortium/connector/codegen/typescript/static/src/ConnectorExporterJsonldStream.ts

@@ -6,13 +6,16 @@ import { Readable } from 'readable-stream';
 
 import IConnectorExporter from "./IConnectorExporter";
 import IConnectorExporterOptions from "./IConnectorExporterOptions";
+import IConnectorExportObserver from "./IConnectorExportObserver";
+import { Observable } from "./observer.js";
 
-export default class ConnectorExporterJsonldStream implements IConnectorExporter {
+export default class ConnectorExporterJsonldStream extends Observable<string> implements IConnectorExporter {
 
     private context?: ContextDefinition;
     private outputContext?: any;
 
     public constructor(context?: ContextDefinition, outputContext?: any) {
+        super();
         this.context = context;
         this.outputContext = outputContext;
     }
@@ -38,7 +41,11 @@ export default class ConnectorExporterJsonldStream implements IConnectorExporter
                 if (outputContext) {
                     json["@context"] = outputContext;
                 }
-                resolve(JSON.stringify(json));
+                const serialized = JSON.stringify(json);
+                this.subscribers.forEach((observer: IConnectorExportObserver) => {
+                    observer.next(serialized);
+                });
+                resolve(serialized);
             });
         });
     }

src/org/datafoodconsortium/connector/codegen/typescript/static/src/ConnectorImporterJsonldStream.ts

@@ -6,14 +6,17 @@ import QuadExt from "rdf-ext/lib/Quad";
 
 import IConnectorImporter from "./IConnectorImporter";
 import IConnectorImporterOptions from "./IConnectorImporterOptions";
+import IConnectorImportObserver from "./IConnectorImportObserver";
+import { Observable } from "./observer.js";
 
-export default class ConnectorImporterJsonldStream implements IConnectorImporter {
+export default class ConnectorImporterJsonldStream extends Observable<DatasetExt[]> implements IConnectorImporter {
 
     private context: string | undefined;
     private documentLoader: any;
 
     // TODO: add the optional parameters of the JsonLdParser class.
     public constructor(parameters?: { context?: any, documentLoader?: any }) {
+        super();
         this.context = parameters?.context;
         this.documentLoader = parameters?.documentLoader;
     }
@@ -112,6 +115,9 @@ export default class ConnectorImporterJsonldStream implements IConnectorImporter
                     datasets.push(blankNodes[0]);
                 }
 
+                this.subscribers.forEach((observer: IConnectorImportObserver) => {
+                    observer.next(datasets);
+                });
                 resolve(datasets);
             });
         });

src/org/datafoodconsortium/connector/codegen/typescript/static/src/IConnectorExportObserver.ts

@@ -0,0 +1,5 @@
+import type { Observer } from './observer';
+
+export default interface IConnectorExportObserver extends Observer<string> {
+  next(json: string): void;
+}

src/org/datafoodconsortium/connector/codegen/typescript/static/src/IConnectorExporter.ts

@@ -1,8 +1,11 @@
 import { Semanticable } from "@virtual-assembly/semantizer";
 import IConnectorExporterOptions from "./IConnectorExporterOptions";
+import { Observable, Subscription } from "./observer";
+import IConnectorExportObserver from "./IConnectorExportObserver";
 
-export default interface IConnectorExporter {
+export default interface IConnectorExporter extends Observable<string> {
 
     export(semanticObjets: Array<Semanticable>, options?: IConnectorExporterOptions): Promise<string>;
-
+
+    subscribe(observer: IConnectorExportObserver): Subscription;
 }

src/org/datafoodconsortium/connector/codegen/typescript/static/src/IConnectorImportObserver.ts

@@ -0,0 +1,6 @@
+import type DatasetExt from "rdf-ext/lib/Dataset";
+import type { Observer } from './observer';
+
+export default interface IConnectorImportObserver extends Observer<DatasetExt[]> {
+  next(datasets: DatasetExt[]): void;
+}

src/org/datafoodconsortium/connector/codegen/typescript/static/src/IConnectorImporter.ts

@@ -1,8 +1,11 @@
 import DatasetExt from "rdf-ext/lib/Dataset";
 import IConnectorImporterOptions from "./IConnectorImporterOptions";
+import { Observable, Subscription } from "./observer";
+import IConnectorImportObserver from "./IConnectorImportObserver";
 
-export default interface IConnectorImporter {
+export default interface IConnectorImporter extends Observable<DatasetExt[]> {
 
     import(data: string, options?: IConnectorImporterOptions): Promise<Array<DatasetExt>>;
-
+
+    subscribe(observer: IConnectorImportObserver): Subscription;
 }