3.0.8

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@push.rocks/smartrx",
-  "version": "3.0.6",
+  "version": "3.0.8",
   "private": false,
   "description": "smart wrapper for rxjs",
   "main": "dist_ts/index.js",
@@ -18,8 +18,8 @@
     "@gitzone/tsbundle": "^2.0.8",
     "@gitzone/tsrun": "^1.2.44",
     "@gitzone/tstest": "^1.0.77",
-    "@push.rocks/tapbundle": "^5.0.12",
+    "@push.rocks/tapbundle": "^5.0.15",
-    "@types/node": "^20.4.4"
+    "@types/node": "^20.8.10"
   },
   "dependencies": {
     "@push.rocks/smartpromise": "^4.0.2",

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@push.rocks/smartrx',
-  version: '3.0.6',
+  version: '3.0.8',
   description: 'smart wrapper for rxjs'
 }

ts/index.ts

@@ -1,5 +1,6 @@
 import * as plugins from './smartrx.plugins.js';
 export * from './smartrx.classes.observablemap.js';
 export * from './smartrx.classes.observableintake.js';
+export * from './smartrx.functions.js';
 import * as rxjs from './smartrx.plugins.rxjs.js';
 export { rxjs };

ts/smartrx.functions.ts

@@ -0,0 +1,32 @@
+import { Observable } from 'rxjs';
+import { Readable } from 'stream';
+
+export function fromStreamWithBackpressure<T>(stream: Readable): Observable<T> {
+  return new Observable<T>((subscriber) => {
+    const pauseStream = () => stream.pause();
+    const resumeStream = () => process.nextTick(() => stream.resume());
+
+    // Handler for each piece of data
+    const onData = (data: T) => {
+      // Pause the stream to apply backpressure
+      pauseStream();
+      // Emit data and resume the stream if the subscriber is ready
+      subscriber.next(data);
+      resumeStream();
+    };
+
+    // Subscribe to stream events
+    stream.on('data', onData);
+    stream.on('error', (error) => subscriber.error(error));
+    stream.on('end', () => subscriber.complete());
+    stream.on('close', () => subscriber.complete());
+
+    // If the subscriber unsubscribes, clean up the stream listeners
+    return () => {
+      stream.removeListener('data', onData);
+      stream.removeListener('error', subscriber.error);
+      stream.removeListener('end', subscriber.complete);
+      stream.removeListener('close', subscriber.complete);
+    };
+  });
+}

ts/smartrx.plugins.rxjs.ts

@@ -1,6 +1,6 @@
 // this file is intended to keep the bundle size down
 
-export { Observable, Subject, fromEvent, ReplaySubject, Subscription, from } from 'rxjs';
+export { Observable, Subject, fromEvent, ReplaySubject, Subscription, from, of } from 'rxjs';
 
 export type { Observer } from 'rxjs';