typedrequest/ts/classes.virtualstream.ts

381 lines
12 KiB
TypeScript
Raw Normal View History

2024-02-21 17:29:35 +00:00
import * as plugins from './plugins.js';
2024-05-30 17:01:39 +00:00
import { TypedRouter } from './classes.typedrouter.js';
2024-02-21 17:29:35 +00:00
export interface ICommFunctions {
sendMethod?: (
2024-02-24 11:20:13 +00:00
sendPayload: plugins.typedRequestInterfaces.IStreamRequest
) => Promise<plugins.typedRequestInterfaces.IStreamRequest>;
2024-02-21 17:29:35 +00:00
typedrouter?: TypedRouter;
}
2024-02-24 11:20:13 +00:00
/**
* 1. A VirtualStream connects over the network
* 2. It is always paired to one other VirtualStream
* on the other side with the same streamId.
* 3. It has a Readable and Writable side.
* 4. The Writable side is Readable on the other side and vice versa.
*/
2024-05-05 15:15:44 +00:00
export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterfaces.IVirtualStream<T> {
2024-02-21 17:29:35 +00:00
// STATIC
public static encodePayloadForNetwork(
objectPayload: any,
2024-02-24 11:20:13 +00:00
commFunctions: ICommFunctions,
originalPayload?: any,
path = []
2024-02-21 17:29:35 +00:00
): any {
2024-02-29 21:51:44 +00:00
if (!objectPayload) {
return objectPayload;
}
2024-02-25 00:54:01 +00:00
if (plugins.smartbuffer.isBufferLike(objectPayload)) {
return objectPayload;
}
2024-02-21 17:29:35 +00:00
if (objectPayload instanceof VirtualStream) {
if (!objectPayload.side && commFunctions.sendMethod) {
objectPayload.side = 'requesting';
objectPayload.sendMethod = commFunctions.sendMethod;
}
if (!objectPayload.side && commFunctions.typedrouter) {
objectPayload.side = 'responding';
objectPayload.typedrouter = commFunctions.typedrouter;
2024-02-24 11:20:13 +00:00
commFunctions.typedrouter.registeredVirtualStreams.add(objectPayload);
}
if (!originalPayload.response || path.includes('response')) {
2024-02-25 00:54:01 +00:00
objectPayload.startKeepAliveLoop();
2024-02-24 11:20:13 +00:00
return {
_isVirtualStream: true,
streamId: objectPayload.streamId,
};
} else {
return {
_OBMITTED_VIRTUAL_STREAM: true,
reason: 'path is under .request: obmitted for deduplication reasons in response cycle.',
};
2024-02-21 17:29:35 +00:00
}
} else if (Array.isArray(objectPayload)) {
2024-02-29 21:59:31 +00:00
// For arrays, we recurse over each item.
return objectPayload.map((item, index) =>
VirtualStream.encodePayloadForNetwork(
item,
commFunctions,
originalPayload || objectPayload,
path.concat(String(index)) // Convert index to string and concatenate to path
)
);
2024-02-21 17:29:35 +00:00
} else if (objectPayload !== null && typeof objectPayload === 'object') {
2024-02-29 21:59:31 +00:00
// For objects, we recurse over each key-value pair.
return Object.entries(objectPayload).reduce((acc, [key, value]) => {
const newPath = path.concat(key); // Concatenate the new key to the path
2024-02-24 11:20:13 +00:00
acc[key] = VirtualStream.encodePayloadForNetwork(
2024-02-29 21:59:31 +00:00
value,
2024-02-24 11:20:13 +00:00
commFunctions,
originalPayload || objectPayload,
2024-02-29 21:59:31 +00:00
newPath
2024-02-24 11:20:13 +00:00
);
2024-02-21 17:29:35 +00:00
return acc;
}, {});
} else {
return objectPayload;
}
}
public static decodePayloadFromNetwork(objectPayload: any, commFunctions: ICommFunctions): any {
2024-05-25 00:15:08 +00:00
if (
plugins.smartbuffer.isBufferLike(objectPayload)
|| objectPayload instanceof TypedRouter
) {
2024-02-25 00:54:01 +00:00
return objectPayload;
}
2024-02-21 17:29:35 +00:00
if (objectPayload !== null && typeof objectPayload === 'object') {
if (objectPayload._isVirtualStream) {
const virtualStream = new VirtualStream();
virtualStream.streamId = objectPayload.streamId;
if (!virtualStream.side && commFunctions.sendMethod) {
virtualStream.side = 'requesting';
virtualStream.sendMethod = commFunctions.sendMethod;
}
if (!virtualStream.side && commFunctions.typedrouter) {
virtualStream.side = 'responding';
virtualStream.typedrouter = commFunctions.typedrouter;
2024-02-24 11:20:13 +00:00
commFunctions.typedrouter.registeredVirtualStreams.add(virtualStream);
2024-02-21 17:29:35 +00:00
}
2024-02-25 00:54:01 +00:00
virtualStream.startKeepAliveLoop();
2024-02-21 17:29:35 +00:00
return virtualStream;
} else if (Array.isArray(objectPayload)) {
const returnArray = [];
for (const item of objectPayload) {
returnArray.push(VirtualStream.decodePayloadFromNetwork(item, commFunctions));
}
return returnArray;
} else {
return Object.keys(objectPayload).reduce((acc, key) => {
acc[key] = VirtualStream.decodePayloadFromNetwork(objectPayload[key], commFunctions);
return acc;
}, {});
}
} else {
return objectPayload;
}
}
// INSTANCE
public side: 'requesting' | 'responding';
public streamId: string = plugins.isounique.uni();
2024-02-24 11:20:13 +00:00
// integration with typedrequest mechanics
public sendMethod: ICommFunctions['sendMethod'];
2024-02-21 17:29:35 +00:00
public typedrouter: TypedRouter;
2024-02-24 11:20:13 +00:00
// wether to keep the stream alive
private keepAlive = true;
2024-02-25 00:54:01 +00:00
private lastKeepAliveEvent: number;
2024-02-24 11:20:13 +00:00
// backpressured arrays
private sendBackpressuredArray =
new plugins.lik.BackpressuredArray<T>(
16
);
private receiveBackpressuredArray =
new plugins.lik.BackpressuredArray<T>(
16
);
2024-02-25 00:54:01 +00:00
constructor() {}
2024-02-24 11:20:13 +00:00
/**
* takes care of sending
*/
private async workOnQueue() {
if(this.side === 'requesting') {
2024-02-29 18:50:25 +00:00
let thisSideIsBackpressured = !this.receiveBackpressuredArray.checkSpaceAvailable();
let otherSideHasNext = false;
let otherSideIsBackpressured = false;
2024-02-24 11:20:13 +00:00
// helper functions
const getFeedback = async () => {
const streamTr = await this.sendMethod({
method: '##VirtualStream##',
request: {
streamId: this.streamId,
cycleId: plugins.isounique.uni(),
cycle: 'request',
mainPurpose: 'feedback',
next: this.sendBackpressuredArray.data.length > 0,
backpressure: this.receiveBackpressuredArray.checkSpaceAvailable(),
},
response: null,
}).catch(() => {
console.log('stream ended immaturely');
this.keepAlive = false;
});
if (streamTr && streamTr.response) {
otherSideIsBackpressured = streamTr.response.backpressure
otherSideHasNext = streamTr.response.next;
}
}
2024-02-29 18:50:25 +00:00
await getFeedback();
2024-02-24 11:20:13 +00:00
// do work loop
while (this.sendBackpressuredArray.data.length > 0 || otherSideHasNext) {
2024-02-25 00:54:01 +00:00
let dataArg: typeof this.sendBackpressuredArray.data[0];
if (this.sendBackpressuredArray.data.length > 0) {
dataArg = this.sendBackpressuredArray.shift();
}
let streamTr: plugins.typedRequestInterfaces.IStreamRequest;
streamTr = await this.sendMethod({
2024-02-24 11:20:13 +00:00
method: '##VirtualStream##',
request: {
streamId: this.streamId,
cycleId: plugins.isounique.uni(),
cycle: 'request',
2024-02-25 00:54:01 +00:00
mainPurpose: dataArg ? 'chunk' : 'read',
2024-02-24 11:20:13 +00:00
backpressure: thisSideIsBackpressured,
next: this.sendBackpressuredArray.data.length > 0,
2024-02-25 00:54:01 +00:00
...dataArg ? { chunkData: dataArg } : {},
2024-02-24 11:20:13 +00:00
},
response: null,
}).catch(() => {
console.log('stream ended immaturely');
this.keepAlive = false;
2024-02-25 00:54:01 +00:00
return null;
2024-02-24 11:20:13 +00:00
});
2024-02-21 17:29:35 +00:00
2024-02-24 11:20:13 +00:00
if (streamTr && streamTr.response && streamTr.response.chunkData) {
this.receiveBackpressuredArray.push(streamTr.response.chunkData);
}
thisSideIsBackpressured = this.receiveBackpressuredArray.checkSpaceAvailable();
// lets care about looping
otherSideHasNext = streamTr && streamTr.response && streamTr.response.next;
}
}
}
/**
* This method handles the stream only on the responding side
* @param streamTrArg
* @returns
*/
2024-02-21 17:29:35 +00:00
public async handleStreamTr(streamTrArg: plugins.typedRequestInterfaces.IStreamRequest) {
2024-02-24 11:20:13 +00:00
if (streamTrArg.request.keepAlive === true && this.keepAlive === true) {
this.lastKeepAliveEvent = Date.now();
} else if (streamTrArg.request.keepAlive === false) {
this.keepAlive = false;
}
// keepAlive handling
if (streamTrArg.request.mainPurpose === 'keepAlive') {
// if the main purpose is keepAlive, we answer with a keepAlive
streamTrArg.response = {
streamId: this.streamId,
cycleId: streamTrArg.request.cycleId,
cycle: 'response',
mainPurpose: 'keepAlive',
keepAlive: this.keepAlive,
next: this.sendBackpressuredArray.data.length > 0,
backpressure: this.receiveBackpressuredArray.checkSpaceAvailable(),
};
}
// feedback handling
if (streamTrArg.request.mainPurpose === 'feedback') {
streamTrArg.response = {
streamId: this.streamId,
cycleId: streamTrArg.request.cycleId,
cycle: 'response',
mainPurpose: 'feedback',
next: this.sendBackpressuredArray.data.length > 0,
backpressure: this.receiveBackpressuredArray.checkSpaceAvailable(),
};
}
// chunk handling
if (streamTrArg.request.mainPurpose === 'chunk') {
this.receiveBackpressuredArray.push(streamTrArg.request.chunkData);
if (this.sendBackpressuredArray.data.length > 0 && streamTrArg.response.backpressure === false) {
const dataArg = this.sendBackpressuredArray.shift();
streamTrArg.response = {
streamId: this.streamId,
cycleId: streamTrArg.request.cycleId,
cycle: 'response',
mainPurpose: 'chunk',
next: this.sendBackpressuredArray.data.length > 1,
backpressure: this.receiveBackpressuredArray.checkSpaceAvailable(),
chunkData: this.sendBackpressuredArray.shift(),
};
} else {
streamTrArg.response = {
streamId: this.streamId,
cycleId: streamTrArg.request.cycleId,
cycle: 'response',
mainPurpose: 'feedback',
next: this.sendBackpressuredArray.data.length > 0,
};
}
streamTrArg.request = null;
}
2024-02-21 17:29:35 +00:00
return streamTrArg;
}
2024-02-25 00:54:01 +00:00
// lifecycle methods
2024-02-21 17:29:35 +00:00
/**
* closes the virtual stream
*/
2024-02-24 11:20:13 +00:00
public async cleanup() {
2024-02-21 17:29:35 +00:00
if (this.typedrouter) {
this.typedrouter.registeredVirtualStreams.remove(this);
}
}
2024-02-24 11:20:13 +00:00
/**
* a keepAlive loop that works across technologies
*/
private async startKeepAliveLoop() {
// initially wait for a second
await plugins.smartdelay.delayFor(0);
let counter = 0;
keepAliveLoop: while (this.keepAlive) {
2024-05-31 20:41:17 +00:00
await this.triggerKeepAlive();
2024-02-24 11:20:13 +00:00
await plugins.smartdelay.delayFor(1000);
}
await plugins.smartdelay.delayFor(1000);
await this.cleanup();
console.log(`cleaned up for stream ${this.streamId}`);
}
private async triggerKeepAlive() {
if (this.side === 'requesting') {
console.log(`keepalive sent.`);
const streamTr = await this.sendMethod({
method: '##VirtualStream##',
request: {
streamId: this.streamId,
cycleId: plugins.isounique.uni(),
cycle: 'request',
mainPurpose: 'keepAlive',
keepAlive: true,
},
response: null,
}).catch(() => {
this.keepAlive = false;
});
// lets handle keepAlive
if (streamTr && streamTr.response && streamTr.response.keepAlive === false) {
this.keepAlive = false;
} else {
this.lastKeepAliveEvent = Date.now();
}
if (streamTr && streamTr.response && streamTr.response.next) {
this.workOnQueue();
}
}
if (Date.now() - this.lastKeepAliveEvent > 10000) {
console.log(`closing stream for ${this.streamId}`);
this.keepAlive = false;
}
}
// Data sending and receiving
public async sendData(dataArg: T): Promise<void> {
this.sendBackpressuredArray.push(dataArg);
this.workOnQueue();
await this.sendBackpressuredArray.waitForSpace();
}
2024-02-25 00:54:01 +00:00
2024-02-24 11:20:13 +00:00
public async fetchData(): Promise<T> {
if (this.receiveBackpressuredArray.hasSpace) {
// do something maybe?
}
await this.receiveBackpressuredArray.waitForItems();
const dataPackage = this.receiveBackpressuredArray.shift();
return dataPackage;
}
2024-02-29 18:50:25 +00:00
/**
* reads from a Readable and sends it to the other side
* @param readableStreamArg
*/
public async readFromWebstream(readableStreamArg: ReadableStream<T>) {
const reader = readableStreamArg.getReader();
let streamIsDone = false;
while(!streamIsDone) {
const { value, done } = await reader.read();
if(value) {
await this.sendData(value);
}
streamIsDone = done;
}
}
public async writeToWebstream(writableStreamArg: WritableStream<T>) {
const writer = writableStreamArg.getWriter();
while(this.keepAlive) {
await writer.write(await this.fetchData());
}
2024-02-24 11:20:13 +00:00
}
2024-02-21 17:29:35 +00:00
}