feat(operators): add JSON serialization and deserialization utilities

Author: StephanGerbeth

.vscode/extensions.json

@@ -15,6 +15,8 @@
     "naumovs.color-highlight",
     "humao.rest-client",
     "techer.open-in-browser",
-    "vitest.explorer"
-  ]
+    "vitest.explorer",
+    "soulcode.vscode-unwanted-extensions"
+  ],
+  "unwantedRecommendations": []
 }

.vscode/settings.json

@@ -48,5 +48,6 @@
   },
   "window.title": "${dirty} ${activeEditorMedium}${separator}${rootName}",
   "workbench.settings.enableNaturalLanguageSearch": false,
-  "task.autoDetect": "off"
+  "task.autoDetect": "off",
+  "terminal.integrated.scrollback": 50000
 }

package-lock.json

@@ -25,12 +25,14 @@
     "test:project:observables": "vitest --project packages/observables",
     "test:project:operators": "vitest --project packages/operators",
     "test:project:playground": "vitest --project packages/playground",
+    "test:project:poc": "vitest --project packages/poc",
     "coverage": "vitest run --coverage"
   },
   "workspaces": [
     "packages/operators",
     "packages/observables",
-    "packages/playground"
+    "packages/playground",
+    "packages/poc"
   ],
   "devDependencies": {
     "@commitlint/config-conventional": "19.6.0",

package.json

@@ -0,0 +1,65 @@
+import { concatAll, concatMap, from, map, Observable, of, toArray } from 'rxjs';
+
+import { createAsyncReplacer, createSyncReplacer } from './json/replacer.js';
+import { createAsyncReviver, createSyncReviver } from './json/reviver.js';
+
+export const serialize = (asyncTransforms, syncTransforms) => source =>
+  source.pipe(
+    traverse(createAsyncReplacer(asyncTransforms)),
+    toJSONString(createSyncReplacer(syncTransforms))
+    //
+  );
+
+export const deserialize = (asyncTransforms, syncTransforms) => source =>
+  source.pipe(
+    fromJSONString(createSyncReviver(syncTransforms)),
+    traverse(createAsyncReviver(asyncTransforms))
+    //
+  );
+
+const traverse = transforms => source =>
+  source.pipe(
+    concatMap(data => of(data).pipe(getOperator(data)(transforms))),
+    transform(transforms)
+    //
+  );
+
+const getOperator = data => traverseInstructions[data.constructor] || (() => source => source);
+
+const traverseInstructions = {
+  [Object]: transforms => source =>
+    source.pipe(
+      map(Object.entries),
+      traverse(transforms),
+      map(Object.fromEntries)
+      //
+    ),
+  [Array]: transforms => source =>
+    source.pipe(
+      concatAll(),
+      traverse(transforms),
+      toArray()
+      //
+    ),
+  [Promise]: transforms => source =>
+    source.pipe(
+      concatMap(value => from(value)),
+      traverse(transforms)
+    ),
+  [Observable]: transforms => source =>
+    source.pipe(
+      concatMap(value => value),
+      traverse(transforms)
+    )
+};
+
+const transform = transforms => source =>
+  source.pipe(concatMap(data => of(data).pipe(findTransform(transforms, data).handler())));
+
+const toJSONString = replacer => source =>
+  source.pipe(map(data => JSON.stringify(data, (_k, v) => findTransform(replacer, v).handler(v))));
+
+const fromJSONString = reviver => source =>
+  source.pipe(map(data => JSON.parse(data, (_k, v) => findTransform(reviver, v).handler(v))));
+
+const findTransform = (transforms, value) => transforms.find(({ validator }) => validator(value));

packages/operators/src/json.js

@@ -0,0 +1,67 @@
+import { Buffer } from 'node:buffer';
+import { readFile } from 'node:fs/promises';
+import { from, lastValueFrom, map, of } from 'rxjs';
+import { TestScheduler } from 'rxjs/testing';
+import { afterAll, beforeEach, describe, expect, test, vi } from 'vitest';
+
+import { deserialize, serialize } from './json';
+
+describe('log', () => {
+  let testScheduler;
+
+  beforeEach(() => {
+    testScheduler = new TestScheduler((actual, expected) => expect(actual).deep.equal(expected));
+  });
+
+  afterAll(() => {
+    vi.restoreAllMocks();
+  });
+
+  test('default', async () => {
+    const replacer = [
+      {
+        validator: value => value?.constructor === Buffer,
+        handler: () => source =>
+          source.pipe(map(buffer => `data:image/jpeg;base64,${buffer.toString('base64')}`))
+      }
+    ];
+
+    const reviver = [
+      {
+        validator: value => value.startsWith && value.startsWith('data:image/jpeg;base64,'),
+        handler: () => source =>
+          source.pipe(
+            map(value => new Blob([Buffer.from(value, 'base64')], { type: 'image/jpeg' }))
+          )
+      }
+    ];
+
+    const data = Promise.resolve({
+      text: Promise.resolve('hello world'),
+      bigInt: BigInt(123),
+      date: new Date(),
+      url: new URL('https://example.com'),
+      image: readFile('./packages/operators/fixtures/images/test_image.jpg'),
+      array: [
+        Promise.resolve('hello world'),
+        BigInt(123),
+        new Date(),
+        new URL('https://example.com')
+      ],
+      nested: Promise.resolve({
+        text: Promise.resolve('hello world'),
+        bigInt: BigInt(123),
+        date: new Date(),
+        url: new URL('https://example.com'),
+        image: from(readFile('./packages/operators/fixtures/images/test_image.jpg'))
+      })
+    });
+    console.log('DATA', data);
+
+    const serialized = await lastValueFrom(of(data).pipe(serialize(replacer)));
+    // console.log('SERIALIZED', serialized);
+
+    const deserialized = await lastValueFrom(of(serialized).pipe(deserialize(reviver)));
+    console.log('DESERIALIZED', deserialized);
+  });
+});

packages/operators/src/json.test.js

@@ -0,0 +1,19 @@
+import { map } from 'rxjs';
+
+const asyncReplacer = [
+  { validator: () => true, handler: () => source => source.pipe(map(value => value)) }
+];
+
+export const syncReplacer = [
+  { validator: value => isURL(value), handler: value => value.toString() },
+  { validator: value => isDate(value), handler: value => value.toISOString() },
+  { validator: value => isBigInt(value), handler: value => value.toString() },
+  { validator: () => true, handler: value => value }
+];
+
+export const createSyncReplacer = (transforms = []) => [...transforms, ...syncReplacer];
+export const createAsyncReplacer = (transforms = []) => [...transforms, ...asyncReplacer];
+
+const isURL = value => value?.constructor === URL;
+const isDate = value => value?.constructor === Date;
+const isBigInt = value => value?.constructor === BigInt;

packages/operators/src/json/replacer.js

@@ -0,0 +1,29 @@
+import { map } from 'rxjs';
+
+const asyncReviver = [
+  { validator: () => true, handler: () => source => source.pipe(map(value => value)) }
+];
+
+export const syncReviver = [
+  { validator: value => isValidUrl(value), handler: value => new URL(value) },
+  { validator: value => isValidISODateString(value), handler: value => new Date(value) },
+  { validator: value => isBigInt(value), handler: value => BigInt(value) },
+  { validator: () => true, handler: value => value }
+];
+
+export const createSyncReviver = (transforms = []) => [...transforms, ...syncReviver];
+export const createAsyncReviver = (transforms = []) => [...transforms, ...asyncReviver];
+
+const isValidUrl = value => {
+  return URL.canParse(value) && /^[\w]+:\/\/\S+$/gm.test(value);
+};
+
+function isValidISODateString(value) {
+  if (!/\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.\d{3}Z/.test(value)) return false;
+  const d = new Date(value);
+  return d instanceof Date && !isNaN(d.getTime()) && d.toISOString() === value; // valid date
+}
+
+function isBigInt(value) {
+  return value?.constructor === String && /^\d+$/.test(value);
+}