Optional metadata for rxjs requesters (#241)

* feat: support optional metadata in rxjs adapter requesters Signed-off-by: Kevin Viglucci <[email protected]> * chore(release): release - rsocket-adapter-rxjs@1.0.0-alpha-rxjs-adapter-optional-metadata.0 - rsocket-composite-metadata@1.0.0-alpha-rxjs-adapter-optional-metadata.0 - [email protected] - [email protected] - [email protected] - [email protected] - [email protected] - rsocket-websocket-client@1.0.0-alpha-rxjs-adapter-optional-metadata.0 - rsocket-websocket-server@1.0.0-alpha-rxjs-adapter-optional-metadata.0 Signed-off-by: Kevin Viglucci <[email protected]>
rsocket · Aug 9, 2022 · f995700 · f995700
1 parent c6d6c47
commit f995700
Show file tree

Hide file tree

Showing 10 changed files with 79 additions and 71 deletions.
diff --git a/packages/rsocket-adapter-rxjs/package.json b/packages/rsocket-adapter-rxjs/package.json
@@ -1,6 +1,6 @@
 {
   "name": "rsocket-adapter-rxjs",
-  "version": "1.0.0-alpha.2",
+  "version": "1.0.0-alpha-rxjs-adapter-optional-metadata.0",
   "license": "Apache-2.0",
   "main": "dist/index",
   "types": "dist/index",
@@ -19,8 +19,8 @@
     "test": "echo \"Error: no test specified\" && exit 0"
   },
   "dependencies": {
-    "rsocket-core": "^1.0.0-alpha.1",
-    "rsocket-messaging": "^1.0.0-alpha.2",
+    "rsocket-core": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0",
+    "rsocket-messaging": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0",
     "rxjs": "^7.4.0"
   },
   "devDependencies": {

diff --git a/packages/rsocket-adapter-rxjs/src/Requesters.ts b/packages/rsocket-adapter-rxjs/src/Requesters.ts
@@ -48,21 +48,23 @@ export function fireAndForget<TData>(
   inputCodec: Codec<TData>
 ): (
   rsocket: RSocket,
-  metadata: Map<string | number | WellKnownMimeType, Buffer>
+  metadata?: Map<string | number | WellKnownMimeType, Buffer>
 ) => Observable<void> {
   return (
     rsocket: RSocket,
-    metadata: Map<string | number | WellKnownMimeType, Buffer>
-  ) =>
-    new RSocketPublisherToObservable((s) =>
+    metadata?: Map<string | number | WellKnownMimeType, Buffer>
+  ) => {
+    const encodedMetadata = metadata ? encodeCompositeMetadata(metadata) : null;
+    return new RSocketPublisherToObservable((s) =>
       rsocket.fireAndForget(
         {
           data: data ? inputCodec.encode(data) : Buffer.allocUnsafe(0),
-          metadata: encodeCompositeMetadata(metadata),
+          metadata: encodedMetadata,
         },
         s
       )
     );
+  };
 }
 
 export function requestResponse<TData, RData>(
@@ -71,23 +73,25 @@ export function requestResponse<TData, RData>(
   outputCodec: Codec<RData>
 ): (
   rsocket: RSocket,
-  metadata: Map<string | number | WellKnownMimeType, Buffer>
+  metadata?: Map<string | number | WellKnownMimeType, Buffer>
 ) => Observable<RData> {
   return (
     rsocket: RSocket,
-    metadata: Map<string | number | WellKnownMimeType, Buffer>
-  ) =>
-    new RSocketPublisherToObservable(
+    metadata?: Map<string | number | WellKnownMimeType, Buffer>
+  ) => {
+    const encodedMetadata = metadata ? encodeCompositeMetadata(metadata) : null;
+    return new RSocketPublisherToObservable(
       (s) =>
         rsocket.requestResponse(
           {
             data: data ? inputCodec.encode(data) : Buffer.allocUnsafe(0),
-            metadata: encodeCompositeMetadata(metadata),
+            metadata: encodedMetadata,
           },
           s
         ),
       outputCodec
     );
+  };
 }
 
 export function requestStream<TData, RData>(
@@ -98,18 +102,19 @@ export function requestStream<TData, RData>(
   scheduler: SchedulerLike = asyncScheduler
 ): (
   rsocket: RSocket,
-  metadata: Map<string | number | WellKnownMimeType, Buffer>
+  metadata?: Map<string | number | WellKnownMimeType, Buffer>
 ) => Observable<RData> {
   return (
     rsocket: RSocket,
-    metadata: Map<string | number | WellKnownMimeType, Buffer>
-  ) =>
-    new RSocketPublisherToPrefetchingObservable(
+    metadata?: Map<string | number | WellKnownMimeType, Buffer>
+  ) => {
+    const encodedMetadata = metadata ? encodeCompositeMetadata(metadata) : null;
+    return new RSocketPublisherToPrefetchingObservable(
       (s, n) =>
         rsocket.requestStream(
           {
             data: data ? inputCodec.encode(data) : Buffer.allocUnsafe(0),
-            metadata: encodeCompositeMetadata(metadata),
+            metadata: encodedMetadata,
           },
           n,
           s
@@ -118,6 +123,7 @@ export function requestStream<TData, RData>(
       outputCodec,
       scheduler
     );
+  };
 }
 
 export function requestChannel<TData, RData>(
@@ -128,9 +134,9 @@ export function requestChannel<TData, RData>(
   scheduler: SchedulerLike = asyncScheduler
 ): (
   rsocket: RSocket,
-  metadata: Map<string | number | WellKnownMimeType, Buffer>
+  metadata?: Map<string | number | WellKnownMimeType, Buffer>
 ) => Observable<RData> {
-  const [firstValueObservable, restValuestObservable] = partition(
+  const [firstValueObservable, restValuesObservable] = partition(
     datas.pipe(
       share({
         connector: () => new Subject(),
@@ -142,35 +148,37 @@ export function requestChannel<TData, RData>(
 
   return (
     rsocket: RSocket,
-    metadata: Map<string | number | WellKnownMimeType, Buffer>
-  ) =>
-    firstValueObservable.pipe(
+    metadata?: Map<string | number | WellKnownMimeType, Buffer>
+  ) => {
+    const encodedMetadata = metadata ? encodeCompositeMetadata(metadata) : null;
+    return firstValueObservable.pipe(
       take(1),
-      concatMap(
-        (firstValue) =>
-          new Observer2BufferingSubscriberToPublisher2PrefetchingObservable(
-            (
-              s: OnTerminalSubscriber &
-                OnNextSubscriber &
-                OnExtensionSubscriber &
-                Requestable &
-                Cancellable
-            ) =>
-              rsocket.requestChannel(
-                {
-                  data: inputCodec.encode(firstValue),
-                  metadata: encodeCompositeMetadata(metadata),
-                },
-                prefetch,
-                false,
-                s
-              ),
-            prefetch,
-            restValuestObservable,
-            inputCodec,
-            outputCodec,
-            scheduler
-          ) as Observable<RData>
-      )
+      concatMap((firstValue) => {
+        return new Observer2BufferingSubscriberToPublisher2PrefetchingObservable(
+          (
+            s: OnTerminalSubscriber &
+              OnNextSubscriber &
+              OnExtensionSubscriber &
+              Requestable &
+              Cancellable
+          ) => {
+            return rsocket.requestChannel(
+              {
+                data: inputCodec.encode(firstValue),
+                metadata: encodedMetadata,
+              },
+              prefetch,
+              false,
+              s
+            );
+          },
+          prefetch,
+          restValuesObservable,
+          inputCodec,
+          outputCodec,
+          scheduler
+        ) as Observable<RData>;
+      })
     );
+  };
 }
diff --git a/packages/rsocket-composite-metadata/package.json b/packages/rsocket-composite-metadata/package.json
@@ -1,6 +1,6 @@
 {
   "name": "rsocket-composite-metadata",
-  "version": "1.0.0-alpha.2",
+  "version": "1.0.0-alpha-rxjs-adapter-optional-metadata.0",
   "license": "Apache-2.0",
   "main": "dist/index",
   "types": "dist/index",
@@ -19,7 +19,7 @@
     "test": "yarn jest"
   },
   "dependencies": {
-    "rsocket-core": "^1.0.0-alpha.1"
+    "rsocket-core": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0"
   },
   "devDependencies": {
     "rimraf": "~3.0.2",

diff --git a/packages/rsocket-core/package.json b/packages/rsocket-core/package.json
@@ -1,6 +1,6 @@
 {
   "name": "rsocket-core",
-  "version": "1.0.0-alpha.1",
+  "version": "1.0.0-alpha-rxjs-adapter-optional-metadata.0",
   "license": "Apache-2.0",
   "main": "dist/index",
   "types": "dist/index",

diff --git a/packages/rsocket-examples/package.json b/packages/rsocket-examples/package.json
@@ -1,6 +1,6 @@
 {
   "name": "rsocket-examples",
-  "version": "1.0.0-alpha.2",
+  "version": "1.0.0-alpha-rxjs-adapter-optional-metadata.0",
   "license": "Apache-2.0",
   "private": true,
   "files": [
@@ -19,13 +19,13 @@
     "start-client-server-rx-composite-metadata-route": "ts-node -r tsconfig-paths/register src/rxjs/ClientServerCompositeMetadataRouteExample.ts"
   },
   "dependencies": {
-    "rsocket-adapter-rxjs": "^1.0.0-alpha.2",
-    "rsocket-composite-metadata": "^1.0.0-alpha.2",
-    "rsocket-core": "^1.0.0-alpha.1",
-    "rsocket-tcp-client": "^1.0.0-alpha.1",
-    "rsocket-tcp-server": "^1.0.0-alpha.1",
-    "rsocket-websocket-client": "^1.0.0-alpha.1",
-    "rsocket-websocket-server": "^1.0.0-alpha.1",
+    "rsocket-adapter-rxjs": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0",
+    "rsocket-composite-metadata": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0",
+    "rsocket-core": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0",
+    "rsocket-tcp-client": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0",
+    "rsocket-tcp-server": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0",
+    "rsocket-websocket-client": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0",
+    "rsocket-websocket-server": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0",
     "ws": "~8.2.3"
   },
   "devDependencies": {

diff --git a/packages/rsocket-messaging/package.json b/packages/rsocket-messaging/package.json
@@ -1,6 +1,6 @@
 {
   "name": "rsocket-messaging",
-  "version": "1.0.0-alpha.2",
+  "version": "1.0.0-alpha-rxjs-adapter-optional-metadata.0",
   "license": "Apache-2.0",
   "main": "dist/index",
   "types": "dist/index",
@@ -19,8 +19,8 @@
     "test": "echo \"Error: no test specified\" && exit 0"
   },
   "dependencies": {
-    "rsocket-composite-metadata": "^1.0.0-alpha.2",
-    "rsocket-core": "^1.0.0-alpha.1"
+    "rsocket-composite-metadata": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0",
+    "rsocket-core": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0"
   },
   "devDependencies": {
     "rimraf": "~3.0.2",

diff --git a/packages/rsocket-tcp-client/package.json b/packages/rsocket-tcp-client/package.json
@@ -1,6 +1,6 @@
 {
   "name": "rsocket-tcp-client",
-  "version": "1.0.0-alpha.1",
+  "version": "1.0.0-alpha-rxjs-adapter-optional-metadata.0",
   "license": "Apache-2.0",
   "main": "dist/index",
   "types": "dist/index",
@@ -20,7 +20,7 @@
     "test": "jest"
   },
   "dependencies": {
-    "rsocket-core": "^1.0.0-alpha.1"
+    "rsocket-core": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0"
   },
   "devDependencies": {
     "rimraf": "~3.0.2",

diff --git a/packages/rsocket-tcp-server/package.json b/packages/rsocket-tcp-server/package.json
@@ -1,6 +1,6 @@
 {
   "name": "rsocket-tcp-server",
-  "version": "1.0.0-alpha.1",
+  "version": "1.0.0-alpha-rxjs-adapter-optional-metadata.0",
   "license": "Apache-2.0",
   "main": "dist/index",
   "types": "dist/index",
@@ -20,7 +20,7 @@
     "test": "jest"
   },
   "dependencies": {
-    "rsocket-core": "^1.0.0-alpha.1"
+    "rsocket-core": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0"
   },
   "devDependencies": {
     "rimraf": "~3.0.2",

diff --git a/packages/rsocket-websocket-client/package.json b/packages/rsocket-websocket-client/package.json
@@ -1,6 +1,6 @@
 {
   "name": "rsocket-websocket-client",
-  "version": "1.0.0-alpha.1",
+  "version": "1.0.0-alpha-rxjs-adapter-optional-metadata.0",
   "license": "Apache-2.0",
   "main": "dist/index",
   "types": "dist/index",
@@ -20,7 +20,7 @@
     "test": "jest"
   },
   "dependencies": {
-    "rsocket-core": "^1.0.0-alpha.1"
+    "rsocket-core": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0"
   },
   "devDependencies": {
     "rimraf": "~3.0.2",

diff --git a/packages/rsocket-websocket-server/package.json b/packages/rsocket-websocket-server/package.json
@@ -1,6 +1,6 @@
 {
   "name": "rsocket-websocket-server",
-  "version": "1.0.0-alpha.1",
+  "version": "1.0.0-alpha-rxjs-adapter-optional-metadata.0",
   "license": "Apache-2.0",
   "main": "dist/index",
   "types": "dist/index",
@@ -20,7 +20,7 @@
     "test": "echo \"Error: no test specified\" && exit 0"
   },
   "dependencies": {
-    "rsocket-core": "^1.0.0-alpha.1",
+    "rsocket-core": "^1.0.0-alpha-rxjs-adapter-optional-metadata.0",
     "ws": "~8.2.3"
   },
   "devDependencies": {