How ot use this with an reader?

[Autumnlight02]

I have a response where I get back a readstream from a fetch call:

const reader = res.response.body!.getReader();
How can I use this with a chain? I tried converting it a few times but failed always...

[uhop]

If you use stream-chain 3.1.0, just include res.response.body as the source stream.

If you are forced to use the older version, just use fromWeb() to adapt ReadableStream to Node's Readable.

[Autumnlight02]

@uhop Issue with that is that the body has a wreid format. I also looked into the json version, and my body is structured as following:

type typeValue
data jsonstring

typeValue has no quotes, type is the key, typeValue is the value, data is a json string which is not always in a chunk, I am trying to get it working with stream-json by creating a custom reader that failed as well.

Also I use stream-chain 3.1.0 and it does not like it at all, perhaps the typescript definitions are outdated?

[Autumnlight02]

This is also related: uhop/stream-json#159

[uhop]

Also I use stream-chain 3.1.0 and it does not like it at all, perhaps the typescript definitions are outdated?

What "typescript definitions"? In any case you are welcome to fix them.

[Autumnlight02]

[just found out I can reply directly:] Are you certain that ReadableStreams are valid? I am testing it, and even if you ignore the TS error it reports that the stream is locked (likely a mistake on my side)

[uhop]

If I understood correctly, you use some custom format (not JSON) and have problems with facilities outside of this library. Is this the correct situation?

[Autumnlight02]

Yeah, the format has a data field which are json chunks, but sometimes a json chunk is too large for a singular chunk, so it gets broken up and then chunk 1 has the first half of the json and chunk 2 the second half, i could solve it without the library using try catch with json.parse but wanted to use a more professional solution which would be this. (Only issue is that I rarely work with streaming and quite struggle, sorry for the inconvinience and thank you for taking your time)

[Autumnlight02]

		return this.direct
			.POST('/api/extra/generate/stream', {
				body: arg.body,
				parseAs: 'stream',
			})
			.then(async (res) => {
				const reader = res.response.body!.getReader();

				const stream = new Readable();
				const getter = chain([
					stream,
					parser(),
					streamValues(),
					// streamArray(),
				]);
				getter.on('data', (data) => {
					// ...
					console.log('data');
				});
				getter.on('end', () => {
					console.log('ennd');
					// ...
				});
				getter.on('error', (err: Error) =>
					console.error('stream failed:', err)
				);

				while (true) {
					const { done, value: valueBuffer } =
						(await reader.read()) as {
							done: Boolean;
							value: Uint8Array;
						};
					if (done) {
						// Do something with last chunk of data then exit reader
						return;
					}
					let res: any[] = [];
					for (let i = 0; i < valueBuffer.length; i++) {
						const arr = new Uint8Array(1);
						arr.set([valueBuffer[i]], 0);
						res.push([valueBuffer[i], textDecoder.decode(arr)]);
					}
					// console.log(res);
					const eventTypeStartInded = 7; // at :
					const eventTypeEndIndex = valueBuffer.indexOf(
						10,
						eventTypeStartInded
					); // find \n
					const eventTypeBuffer = valueBuffer.slice(
						eventTypeStartInded,
						eventTypeEndIndex
					);
					const eventType = textDecoder.decode(eventTypeBuffer);
					// " " (space) is 32
					//Jump over data: text directly to json start
					const dataStart = eventTypeEndIndex + 7;
					const dataEnd = valueBuffer.length;
					const dataBuffer = valueBuffer.slice(dataStart, dataEnd);
					const dataString = textDecoder.decode(dataBuffer);
					// const eventType = string.slice(
					// 	startEventType,
					// 	endEventType
					// );
					console.log(dataString);
					stream.push(dataString);
					const result = {
						eventType: eventType,
					};
					// Otherwise do something here to process current chunk
				}
			});

Basically this is the code I am trying to run, and my goal is to either use chain to only read the json or all the data, I am quite new to the library and find it a bit difficult to use.

Regarding the typescript issue, If you state that compability is a Thing I will push a fix for that latest tommorow, the error state is the following:

[{
	"resource": "/var/home/autumnlight/Documents/github/personal/koboldcpp/src/Koboldcpp.ts",
	"owner": "typescript",
	"code": "2322",
	"severity": 8,
	"message": "Type 'ReadableStream<any> | null' is not assignable to type 'StreamItem'.\n  Type 'ReadableStream<any>' is not assignable to type 'StreamItem'.\n    Type 'ReadableStream<any>' is missing the following properties from type 'Transform': _transform, _flush, writable, writableEnded, and 69 more.",
	"source": "ts",
	"startLineNumber": 76,
	"startColumn": 6,
	"endLineNumber": 76,
	"endColumn": 23
}]

[uhop]

Could you provide a representative (simulated, as small as possible) sample of data you read? Otherwise, it is difficult to understand the code.

[Autumnlight02]

raw output below:

-- chunk console log: --
event: message
data: {"token": " generals generals generals generals Battle generals Battle Battle generals generals Battle Battle generals generals generals generals generals Battle Battle Battle generals generals Battle generals generals generals Battle Battle generals generals generals generals Battle generals generals generals Battle generals Battle generals generals generals generals generals generals generals generals generals generals generals generals Battle Battle Battle generals generals generals generals Battle generals generals Battle Battle generals generals generals generals Battle Battle generals generals generals Battle generals generals generals Battle generals generals generals generals Battle", "finish_reason": "null"}

event: message
data: {"token": " Battle", "fini
-- chunk console log: --
sh_reason": "null"}


-- chunk console log: --
event: message
data: {"token": " generals generals generals Battle generals Battle Battle", "finish_reason": "null"}


-- chunk console log: --
event: message
data: {"token": " Battle generals generals generals Battle Battle generals generals", "finish_reason": "null"}


-- chunk console log: --
event: message
data: {"token": " generals Battle generals generals generals Battle generals", "finish_reason": "null"}

event: message
data: {"token": " generals", "finish_reason": "null"}


-- chunk console log: --
event: message
data: {"token": " Battle Battle generals generals Battle generals generals", "finish_reason": "null"}


-- chunk console log: --
event: message
data: {"token": " generals generals generals generals generals generals Battle generals", "finish_reason": "null"}


-- chunk console log: --
event: message
data: {"token": " generals Battle Battle generals Battle Battle Battle", "finish_reason": "null"}


-- chunk console log: --
event: message
data: {"token": " Battle generals generals generals Battle Battle Battle generals", "finish_reason": "null"}


-- chunk console log: --
event: message
data: {"token": " Battle Battle generals generals generals Battle generals", "finish_reason": "null"}


-- chunk console log: --
event: message
data: {"token": " Battle generals Battle Battle generals Battle Battle", "finish_reason": "length"}

[Autumnlight02]

I just noticed, the broken up json has no -- chunk console log --

[Autumnlight02]

another example:

-- chunk console log: --           
event: message
data: {"token": " generals Battle generals generals Battle Battle generals", "finish_reason": "null"}


-- chunk console log: --           
event: message
data: {"token": " Battle generals Battle generals Battle Battle generals Battle", "finish_reason": "null"}
-- chunk console log: --           
event: message
data: {"token": " Battle Battle generals generals generals Battle generals Battle Battle generals generals generals generals generals generals Battle generals Battle Battle generals generals Battle generals generals generals generals generals Battle generals generals Battle generals generals Battle generals generals generals Battle Battle Battle generals Battle generals Battle generals generals generals Battle generals Battle generals generals generals generals generals Battle generals Battle Battle generals generals Battle Battle generals Battle generals generals generals generals generals generals generals Battle generals generals generals generals Battle Battle generals Battle generals generals generals generals generals Battle generals Battle generals Battle Battle g
-- chunk console log: --           
enerals Battle Battle generals Battle generals", "finish_reason": "null"}

[Autumnlight02]

I found a solution without stream chain, thank you for your help