chore: add rxjs example without rsocket-messaging

Signed-off-by: Kevin Viglucci <[email protected]>

Author: viglucci

packages/rsocket-examples/package.json

@@ -16,6 +16,7 @@
     "start-client-server-request-response-tcp": "ts-node -r tsconfig-paths/register src/ClientServerRequestResponseExampleTcp.ts",
     "start-client-server-request-response-websocket": "ts-node -r tsconfig-paths/register src/ClientServerRequestResponseExampleWebSocket.ts",
     "start-client-server-composite-metadata-route": "ts-node -r tsconfig-paths/register src/ClientServerCompositeMetadataRouteExample.ts",
+    "start-client-server-rxjs-requester-responder": "ts-node -r tsconfig-paths/register src/rxjs/RxjsRequesterResponderExample.ts",
     "start-client-server-rxjs-messaging-composite-metadata-route": "ts-node -r tsconfig-paths/register src/rxjs/RxjsMessagingCompositeMetadataRouteExample.ts",
     "start-client-apollo-graphql": "ts-node -r tsconfig-paths/register src/graphql/apollo/client/example.ts",
     "start-client-server-apollo-graphql": "ts-node -r tsconfig-paths/register src/graphql/apollo/client-server/example.ts"

packages/rsocket-examples/src/rxjs/RxjsRequesterResponderExample.ts

@@ -0,0 +1,229 @@
+/*
+ * Copyright 2021-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { RSocket, RSocketConnector, RSocketServer } from "rsocket-core";
+import { Codec } from "rsocket-messaging";
+import { RxRequestersFactory, RxRespondersFactory } from "rsocket-adapter-rxjs";
+import { TcpClientTransport } from "rsocket-tcp-client";
+import { TcpServerTransport } from "rsocket-tcp-server";
+import { exit } from "process";
+import {
+  firstValueFrom,
+  interval,
+  map,
+  Observable,
+  of,
+  take,
+  tap,
+  timer,
+} from "rxjs";
+import Logger from "../shared/logger";
+
+let serverCloseable;
+
+class EchoService {
+  handleEchoRequestResponse(data: string): Observable<string> {
+    return timer(1000).pipe(map(() => `Echo: ${data}`));
+  }
+
+  handleEchoRequestStream(data: string): Observable<string> {
+    return interval(1000).pipe(
+      map(() => `RxEchoService Echo: ${data}`),
+      tap((value) => {
+        Logger.info(`[server] sending: ${value}`);
+      })
+    );
+  }
+
+  handleEchoRequestChannel(datas: Observable<string>): Observable<string> {
+    datas
+      .pipe(
+        tap((value) => {
+          Logger.info(`[server] receiving: ${value}`);
+        })
+      )
+      .subscribe();
+    return interval(200).pipe(
+      map((data) => `RxEchoService Echo: ${data}`),
+      tap((value) => {
+        Logger.info(`[server] sending: ${value}`);
+      })
+    );
+  }
+
+  handleLogFireAndForget(data: string): Observable<void> {
+    Logger.info(`[server] received: ${data}`);
+    return of(null);
+  }
+}
+
+function makeServer() {
+  const stringCodec = new StringCodec();
+  return new RSocketServer({
+    transport: new TcpServerTransport({
+      listenOptions: {
+        port: 9090,
+        host: "127.0.0.1",
+      },
+    }),
+    acceptor: {
+      accept: async () => {
+        const echoService = new EchoService();
+        return {
+          fireAndForget: RxRespondersFactory.fireAndForget(
+            echoService.handleLogFireAndForget,
+            stringCodec
+          ),
+          requestResponse: RxRespondersFactory.requestResponse(
+            echoService.handleEchoRequestResponse,
+            { inputCodec: stringCodec, outputCodec: stringCodec }
+          ),
+          requestStream: RxRespondersFactory.requestStream(
+            echoService.handleEchoRequestStream,
+            { inputCodec: stringCodec, outputCodec: stringCodec }
+          ),
+          requestChannel: RxRespondersFactory.requestChannel(
+            echoService.handleEchoRequestChannel,
+            { inputCodec: stringCodec, outputCodec: stringCodec },
+            4
+          ),
+        };
+      },
+    },
+  });
+}
+
+function makeConnector() {
+  return new RSocketConnector({
+    transport: new TcpClientTransport({
+      connectionOptions: {
+        host: "127.0.0.1",
+        port: 9090,
+      },
+    }),
+  });
+}
+
+async function fireAndForget(rsocket: RSocket, route: string = "") {
+  return new Promise((resolve, reject) => {
+    const request = RxRequestersFactory.fireAndForget(
+      "Hello World",
+      stringCodec
+    );
+    request(rsocket).subscribe({
+      complete() {
+        // give server a chance to handle before continuing
+        setTimeout(() => {
+          resolve(null);
+        }, 100);
+      },
+      error(err) {
+        reject(err);
+      },
+    });
+  });
+}
+
+async function requestResponse(rsocket: RSocket) {
+  const request = RxRequestersFactory.requestResponse(
+    "Hello World",
+    stringCodec,
+    stringCodec
+  );
+  return firstValueFrom(
+    request(rsocket).pipe(tap((data) => Logger.info(`payload[data: ${data};]`)))
+  );
+}
+
+async function requestStream(rsocket: RSocket) {
+  const request = RxRequestersFactory.requestStream(
+    "Hello World",
+    stringCodec,
+    stringCodec,
+    5
+  );
+  return request(rsocket)
+    .pipe(
+      tap((data) => {
+        Logger.info(`[client] received[data: ${data}]`);
+      }),
+      take(10)
+    )
+    .toPromise();
+}
+
+async function requestChannel(rsocket: RSocket) {
+  const request = RxRequestersFactory.requestChannel(
+    interval(1000).pipe(
+      map((i) => `Hello World ${i}`),
+      tap((data) => {
+        Logger.info(`[client] produced[data: ${data}]`);
+      })
+    ),
+    stringCodec,
+    stringCodec,
+    5
+  );
+  return request(rsocket)
+    .pipe(
+      tap((data) => {
+        Logger.info(`[client] received[data: ${data}]`);
+      }),
+      take(25)
+    )
+    .toPromise();
+}
+
+class StringCodec implements Codec<string> {
+  readonly mimeType: string = "text/plain";
+
+  decode(buffer: Buffer): string {
+    return buffer.toString();
+  }
+
+  encode(entity: string): Buffer {
+    return Buffer.from(entity);
+  }
+}
+
+const stringCodec = new StringCodec();
+
+async function main() {
+  const server = makeServer();
+  const connector = makeConnector();
+
+  serverCloseable = await server.bind();
+  const rsocket = await connector.connect();
+
+  Logger.info(`----- Fire And Forget -----`);
+  await fireAndForget(rsocket);
+
+  Logger.info(`----- Request Response -----`);
+  await requestResponse(rsocket);
+
+  Logger.info(`----- Request Stream -----`);
+  await requestStream(rsocket);
+
+  Logger.info(`----- Request Channel -----`);
+  await requestChannel(rsocket);
+}
+
+main()
+  .then(() => exit())
+  .catch((error: Error) => {
+    console.error(error);
+    exit(1);
+  });