feat(js): Refacto socket handling for big messages, add examples, fix singleConsumerStream

foreign/node/examples/stream-file-to-topic.ts

@@ -0,0 +1,101 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 { HeaderValue } from '../src/index.js';
+import { open } from 'node:fs/promises';
+import { resolve } from 'node:path';
+import { pipeline } from 'node:stream/promises';
+import { Writable, type TransformCallback } from 'node:stream';
+import { getClient, ensureStream, ensureTopic } from './utils.js';
+
+export const fileToTopic = async (
+  filepath: string,
+  streamId: number,
+  topicId: number,
+  highWaterMark = 512 * 1024
+) => {
+  const cli = getClient();
+  await ensureStream(cli, streamId);
+  await ensureTopic(cli, streamId, topicId);
+  const fd = await open(filepath);
+  const fname = filepath.split('/').pop() || filepath;
+  const st = await fd.stat();
+  console.log('FILE/STAT', fname, '~', (st.size / (1024 * 1024)).toFixed(2), 'mb', st);
+  const dStart = Date.now();
+
+  try {
+    let idx = 0;
+
+    await pipeline(
+      fd.createReadStream({highWaterMark}),
+
+      new Writable({
+        async write(chunks: Buffer, encoding: BufferEncoding, cb: TransformCallback) {
+          const messages = [{
+            headers: {
+              fileindex: HeaderValue.Uint32(idx++),
+              filename: HeaderValue.String(fname)
+            },
+            payload: chunks
+          }];
+
+          try {
+            await cli.message.send({ streamId, topicId, messages });
+            cb();
+          } catch (err) {
+            console.log('WRITE ERR', err, chunks);
+            cb(err as Error);
+          }
+        }
+      })
+    );
+
+    console.log(`Finished ! took ${Date.now() - dStart}ms`,);
+  } catch (err) {
+    console.error('Pipeline failed !', err);
+    throw err;
+  }
+};
+
+
+const argz = process.argv.slice(2);
+const [rPath, streamIdStr, topicIdStr] = argz;
+
+if (argz.length < 3 || ['-h', '--help', '?'].includes(argz[0])) {
+  console.log(`Usage: node stream-file-to-topic.js filePath streamId topicId`)
+  console.log('got', argz);
+  console.log('note: this script only accept numerical stream/topic id');
+  process.exit(1);
+}
+
+const filepath = resolve(rPath);
+const streamId = parseInt(streamIdStr);
+const topicId = parseInt(topicIdStr);
+
+console.log('running with params:', { filepath, streamId, topicId })
+
+try {
+  await fileToTopic(filepath, streamId, topicId, 512 * 1024);
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
+} catch(err) {
+  process.exit(1);
+}
+
+process.exit(0);

foreign/node/examples/stream-topic-to-file.ts

@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 { PollingStrategy, singleConsumerStream, Consumer } from '../src/index.js';
+import { open } from 'node:fs/promises';
+import { resolve } from 'node:path';
+import { getClient } from './utils.js';
+import type { PollMessagesResponse } from '../src/wire/index.js';
+
+
+export const topicToFile = async (
+  filepath: string,
+  streamId: number,
+  topicId: number,
+  partitionId = 1
+) => {
+  const cli = getClient();
+  const fd = await open(filepath, 'w+');
+
+  const t = await cli.topic.get({ streamId, topicId });
+  console.log('TOPIC/GET', t);
+
+  // reset consumer offset 
+  try {
+    const offset = await cli.offset.get({
+      streamId, topicId, partitionId, consumer: Consumer.Single
+    });
+    if (offset)
+      await cli.offset.delete({ streamId, topicId, partitionId, consumer: Consumer.Single });
+    // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  } catch (err) {}
+
+  const dStart = Date.now();
+  const consumer = singleConsumerStream(cli._config);
+  const stream = await consumer({
+    streamId,
+    topicId,
+    partitionId,
+    pollingStrategy: PollingStrategy.Next,
+    count: 1,
+    interval: 1000,
+    endOnLastOffset: true
+  });
+
+  stream.on('data', async (polled: PollMessagesResponse) => {
+    const { messages } = polled;
+    return await fd.write(Buffer.concat(messages.map(m => m.payload)));
+  });
+
+  stream.once('end', () => {
+    console.log('pollStream#end', `took ${Date.now() - dStart}ms`);
+    fd.datasync();
+    fd.close();
+    stream.destroy();
+  });
+
+  stream.on('error', (err) => {
+    console.log('pollStream#err', err);
+    stream.destroy();
+  });
+
+  return stream;
+};
+
+
+const argz = process.argv.slice(2);
+const [rPath, streamIdStr, topicIdStr] = argz;
+
+if (argz.length < 3 || ['-h', '--help', '?'].includes(argz[0])) {
+  console.log(`Usage: node stream-file-to-topic.js filePath streamId topicId`)
+  console.log('got', argz);
+  console.log('note: this script only accept numerical stream/topic id');
+  process.exit(1);
+}
+
+const filepath = resolve(rPath);
+const streamId = parseInt(streamIdStr);
+const topicId = parseInt(topicIdStr);
+
+console.log('running with params:', { filepath, streamId, topicId });
+
+const s = await topicToFile(filepath, streamId, topicId);
+
+s.on('end', () => {
+  console.log('end, exiting');
+  process.exit(0);
+});
+
+s.on('error', (err) => {
+  console.error('FAILED WITH ERROR', err, 'exiting...');
+  process.exit(1);
+});

foreign/node/examples/utils.ts

@@ -0,0 +1,67 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 { Client } from '../src/index.js';
+import { getIggyAddress } from '../src/tcp.sm.utils.js';
+
+
+export const getClient = () => {
+  const [host, port] = getIggyAddress();
+  const credentials = { username: 'iggy', password: 'iggy' };
+
+  const opt = {
+    transport: 'TCP' as const,
+    options: { host, port },
+    credentials
+  };
+
+  return new Client(opt);
+}
+
+export const ensureStream = async (cli: Client, streamId: number) => {
+  try {
+    return !!await cli.stream.get({ streamId });
+    // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  } catch (err) {
+    return cli.stream.create({ streamId, name: `ensure-stream-${streamId}` })
+  }
+}
+
+export const ensureTopic = async (
+  cli: Client,
+  streamId: number,
+  topicId: number,
+  partitionCount = 1,
+  compressionAlgorithm = 1
+) => {
+  try {
+    return !!await cli.topic.get({ streamId, topicId });
+    // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  } catch (err) {
+    return cli.topic.create({
+      streamId,
+      topicId,
+      name: `ensure-topic-${streamId}-${topicId}`,
+      partitionCount,
+      compressionAlgorithm
+    });
+  }
+}
+

foreign/node/package.json

@@ -15,14 +15,17 @@
   },
   "publishConfig": {
     "access": "public",
-    "main": "dist/index.js",
-    "types": "dist/index.d.ts"
+    "main": "dist/src/index.js",
+    "types": "dist/src/index.d.ts"
   },
   "homepage": "https://iggy.apache.org",
   "files": [
-    "dist/**/*.js",
-    "dist/**/*.d.ts",
-    "dist/index.d.ts"
+    "dist/src/**/*.js",
+    "dist/src/**/*.d.ts",
+    "dist/src/index.d.ts",
+    "!dist/src/bdd/",
+    "!dist/src/e2e/",
+    "!dist/**/*.test.*"
   ],
   "types": "dist/index.d.ts",
   "main": "dist/index.js",
@@ -32,7 +35,7 @@
     "test:bdd": "cucumber-js --exit",
     "test": "npm run test:unit && npm run test:bdd && npm run test:e2e",
     "clean": "rm -Rf dist/",
-    "lint": "eslint src/**/*.ts",
+    "lint": "eslint src/**/*.ts examples/**/*.ts",
     "commitlint": "commitlint --edit",
     "build": "tsc -p tsconfig.json",
     "start": "node dist/index.js",

foreign/node/src/bdd/message.ts

@@ -20,7 +20,7 @@
 
 import assert from 'node:assert/strict';
 import { When, Then } from "@cucumber/cucumber";
-import { ConsumerKind, PollingStrategy, Partitioning } from '../wire/index.js';
+import { Consumer,  PollingStrategy, Partitioning } from '../wire/index.js';
 import { someMessageContent } from '../tcp.sm.utils.js';
 import type { TestWorld } from './world.js';
 
@@ -67,7 +67,7 @@ When(
     const pollReq = {
       streamId,
       topicId,
-      consumer: { kind: ConsumerKind.Single, id: 1 },
+      consumer: Consumer.Single,
       partitionId,
       pollingStrategy: PollingStrategy.Offset(BigInt(offset)),
       count: 100,