Skip to content

Commit 48d122a

Browse files
viglucciviglucci
authored
chore: add additional examples (#258)
* chore: add nvmrc file * chore: add tcp request stream example * chore: add client-only composite metadata routing example * chore: fix comment typos in various examples
1 parent 8f862ad commit 48d122a

File tree

5 files changed

+309
-1
lines changed

5 files changed

+309
-1
lines changed

‎.nvmrc

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
v18.12.1

‎packages/rsocket-examples/package.json

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,8 @@
1212
"start-client-request-channel": "ts-node -r tsconfig-paths/register src/ClientRequestChannelExample.ts",
1313
"start-client-server-request-channel-resume": "ts-node -r tsconfig-paths/register src/ClientServerRequestChannelResumeExample.ts",
1414
"start-client-request-fnf-with-lease": "ts-node -r tsconfig-paths/register src/ClienRequestFnfnWithLeaseExampleTcp.ts",
15+
"start-client-composite-metadata-route": "ts-node -r tsconfig-paths/register src/ClientCompositeMetadataRouteExample.ts",
16+
"start-client-server-request-stream-tcp": "ts-node -r tsconfig-paths/register src/ClientServerRequestStreamExampleTCP.ts",
1517
"start-client-server-request-stream-websocket": "ts-node -r tsconfig-paths/register src/ClientServerRequestStreamExampleWebSocket.ts",
1618
"start-client-server-request-response-tcp": "ts-node -r tsconfig-paths/register src/ClientServerRequestResponseExampleTcp.ts",
1719
"start-client-server-request-response-websocket": "ts-node -r tsconfig-paths/register src/ClientServerRequestResponseExampleWebSocket.ts",

‎packages/rsocket-examples/src/ClientCompositeMetadataRouteExample.ts

Lines changed: 138 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,138 @@
1+
import { RSocket, RSocketConnector } from "@rsocket/core";
2+
import { TcpClientTransport } from "@rsocket/transport-tcp-client";
3+
import {
4+
encodeCompositeMetadata,
5+
encodeRoute,
6+
WellKnownMimeType,
7+
} from "@rsocket/composite-metadata";
8+
import Logger from "./shared/logger";
9+
import { exit } from "process";
10+
import MESSAGE_RSOCKET_ROUTING = WellKnownMimeType.MESSAGE_RSOCKET_ROUTING;
11+
import MESSAGE_RSOCKET_COMPOSITE_METADATA = WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA;
12+
13+
/**
14+
* This example assumes you have a RSocket server running on 127.0.0.1:9000 that will respond
15+
* to requests at the following routes:
16+
* - login (requestResponse)
17+
* - message (requestResponse)
18+
* - messages.incoming (requestStream)
19+
*/
20+
21+
function makeConnector() {
22+
const connectorConnectionOptions = {
23+
host: "127.0.0.1",
24+
port: 9000,
25+
};
26+
console.log(
27+
`Creating connector to ${JSON.stringify(connectorConnectionOptions)}`
28+
);
29+
return new RSocketConnector({
30+
setup: {
31+
metadataMimeType: MESSAGE_RSOCKET_COMPOSITE_METADATA.string,
32+
},
33+
transport: new TcpClientTransport({
34+
connectionOptions: connectorConnectionOptions,
35+
}),
36+
});
37+
}
38+
39+
function createRoute(route?: string) {
40+
let compositeMetaData = undefined;
41+
if (route) {
42+
const encodedRoute = encodeRoute(route);
43+
44+
const map = new Map<WellKnownMimeType, Buffer>();
45+
map.set(MESSAGE_RSOCKET_ROUTING, encodedRoute);
46+
compositeMetaData = encodeCompositeMetadata(map);
47+
}
48+
return compositeMetaData;
49+
}
50+
51+
async function requestResponse(rsocket: RSocket, route: string, data: string) {
52+
console.log(`Executing requestResponse: ${JSON.stringify({ route, data })}`);
53+
return new Promise((resolve, reject) => {
54+
return rsocket.requestResponse(
55+
{
56+
data: Buffer.from(data),
57+
metadata: createRoute(route),
58+
},
59+
{
60+
onError: (e) => {
61+
reject(e);
62+
},
63+
onNext: (payload, isComplete) => {
64+
Logger.info(
65+
`requestResponse onNext payload[data: ${payload.data}; metadata: ${payload.metadata}]|${isComplete}`
66+
);
67+
resolve(payload);
68+
},
69+
onComplete: () => {
70+
Logger.info(`requestResponse onComplete`);
71+
resolve(null);
72+
},
73+
onExtension: () => {},
74+
}
75+
);
76+
});
77+
}
78+
79+
async function main() {
80+
const connector = makeConnector();
81+
const rsocket = await connector.connect();
82+
83+
await requestResponse(rsocket, "login", "user1");
84+
85+
await requestResponse(
86+
rsocket,
87+
"message",
88+
JSON.stringify({ user: "user1", content: "a message" })
89+
);
90+
91+
await new Promise((resolve, reject) => {
92+
let payloadsReceived = 0;
93+
const maxPayloads = 10;
94+
const requester = rsocket.requestStream(
95+
{
96+
data: Buffer.from("Hello World"),
97+
metadata: createRoute("messages.incoming"),
98+
},
99+
3,
100+
{
101+
onError: (e) => reject(e),
102+
onNext: (payload, isComplete) => {
103+
Logger.info(
104+
`[client] payload[data: ${payload.data}; metadata: ${payload.metadata}]|isComplete: ${isComplete}`
105+
);
106+
107+
payloadsReceived++;
108+
109+
// request 5 more payloads every 5th payload, until a max total payloads received
110+
if (payloadsReceived % 2 == 0 && payloadsReceived < maxPayloads) {
111+
requester.request(2);
112+
} else if (payloadsReceived >= maxPayloads) {
113+
requester.cancel();
114+
setTimeout(() => {
115+
resolve(null);
116+
});
117+
}
118+
119+
if (isComplete) {
120+
resolve(null);
121+
}
122+
},
123+
onComplete: () => {
124+
Logger.info(`requestStream onComplete`);
125+
resolve(null);
126+
},
127+
onExtension: () => {},
128+
}
129+
);
130+
});
131+
}
132+
133+
main()
134+
.then(() => exit())
135+
.catch((error: Error) => {
136+
console.error(error);
137+
exit(1);
138+
});

‎packages/rsocket-examples/src/ClientServerRequestStreamExampleTCP.ts

Lines changed: 167 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,167 @@
1+
/*
2+
* Copyright 2021-2022 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
import {
18+
OnExtensionSubscriber,
19+
OnNextSubscriber,
20+
OnTerminalSubscriber,
21+
Payload,
22+
RSocketConnector,
23+
RSocketServer,
24+
} from "@rsocket/core";
25+
import { WebsocketClientTransport } from "@rsocket/transport-websocket-client";
26+
import { WebsocketServerTransport } from "@rsocket/transport-websocket-server";
27+
import { exit } from "process";
28+
import WebSocket from "ws";
29+
import Logger from "./shared/logger";
30+
import { TcpServerTransport } from "@rsocket/transport-tcp-server";
31+
import { TcpClientTransport } from "@rsocket/transport-tcp-client";
32+
33+
function makeServer() {
34+
return new RSocketServer({
35+
transport: new TcpServerTransport({
36+
listenOptions: {
37+
port: 9090,
38+
host: "127.0.0.1",
39+
},
40+
}),
41+
acceptor: {
42+
accept: async () => {
43+
return {
44+
requestStream: (
45+
payload: Payload,
46+
initialRequestN,
47+
responderStream: OnTerminalSubscriber &
48+
OnNextSubscriber &
49+
OnExtensionSubscriber
50+
) => {
51+
Logger.info(
52+
`[server] requestStream payload[data: ${payload.data}; metadata: ${payload.metadata}]|initialRequestN: ${initialRequestN}`
53+
);
54+
55+
let interval = null;
56+
let requestedResponses = initialRequestN;
57+
let sentResponses = 0;
58+
59+
// simulate async data with interval
60+
interval = setInterval(() => {
61+
sentResponses++;
62+
let isComplete = sentResponses >= requestedResponses;
63+
responderStream.onNext(
64+
{
65+
data: Buffer.from(new Date()),
66+
metadata: undefined,
67+
},
68+
isComplete
69+
);
70+
if (isComplete) {
71+
clearInterval(interval);
72+
}
73+
}, 750);
74+
75+
return {
76+
cancel() {
77+
Logger.info("[server] stream cancelled by client");
78+
clearInterval(interval);
79+
},
80+
request(n) {
81+
requestedResponses += n;
82+
Logger.info(
83+
`[server] request n: ${n}, requestedResponses: ${requestedResponses}, sentResponses: ${sentResponses}`
84+
);
85+
},
86+
onExtension: () => {},
87+
};
88+
},
89+
};
90+
},
91+
},
92+
});
93+
}
94+
95+
function makeConnector() {
96+
const connectorConnectionOptions = {
97+
host: "127.0.0.1",
98+
port: 9090,
99+
};
100+
console.log(
101+
`Creating connector to ${JSON.stringify(connectorConnectionOptions)}`
102+
);
103+
return new RSocketConnector({
104+
transport: new TcpClientTransport({
105+
connectionOptions: connectorConnectionOptions,
106+
}),
107+
});
108+
}
109+
110+
let serverCloseable;
111+
112+
async function main() {
113+
const server = makeServer();
114+
const connector = makeConnector();
115+
116+
serverCloseable = await server.bind();
117+
const rsocket = await connector.connect();
118+
119+
await new Promise((resolve, reject) => {
120+
let payloadsReceived = 0;
121+
const maxPayloads = 10;
122+
const requester = rsocket.requestStream(
123+
{
124+
data: Buffer.from("Hello World"),
125+
},
126+
3,
127+
{
128+
onError: (e) => reject(e),
129+
onNext: (payload, isComplete) => {
130+
Logger.info(
131+
`[client] payload[data: ${payload.data}; metadata: ${payload.metadata}]|isComplete: ${isComplete}`
132+
);
133+
134+
payloadsReceived++;
135+
136+
// request 5 more payloads every 5th payload, until a max total payloads received
137+
if (payloadsReceived % 2 == 0 && payloadsReceived < maxPayloads) {
138+
requester.request(2);
139+
} else if (payloadsReceived >= maxPayloads) {
140+
requester.cancel();
141+
setTimeout(() => {
142+
resolve(null);
143+
});
144+
}
145+
146+
if (isComplete) {
147+
resolve(null);
148+
}
149+
},
150+
onComplete: () => {
151+
resolve(null);
152+
},
153+
onExtension: () => {},
154+
}
155+
);
156+
});
157+
}
158+
159+
main()
160+
.then(() => exit())
161+
.catch((error: Error) => {
162+
console.error(error);
163+
exit(1);
164+
})
165+
.finally(() => {
166+
serverCloseable?.close();
167+
});

‎packages/rsocket-examples/src/ClientServerRequestStreamExampleWebSocket.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -126,7 +126,7 @@ async function main() {
126126

127127
payloadsReceived++;
128128

129-
// request 5 more payloads event 5th payload, until a max total payloads received
129+
// request 5 more payloads every 5th payload, until a max total payloads received
130130
if (payloadsReceived % 2 == 0 && payloadsReceived < maxPayloads) {
131131
requester.request(2);
132132
} else if (payloadsReceived >= maxPayloads) {

0 commit comments

Comments
 (0)