foreign/node/src/tcp.sm.utils.ts

/**
 * 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 assert from "node:assert/strict";
import { v7 } from "./wire/uuid.utils.js";
import {
  sendMessages,
  type Partitioning,
  HeaderValue,
  HeaderKeyFactory,
  type Message,
} from "./wire/index.js";
import type { ClientProvider } from "./client/client.type.js";
import type { Id } from "./wire/identifier.utils.js";

const h0 = [
  { key: HeaderKeyFactory.String("foo"), value: HeaderValue.Int32(42) },
  { key: HeaderKeyFactory.String("bar"), value: HeaderValue.Uint8(123) },
];
const h1 = [
  {
    key: HeaderKeyFactory.String("x-header-string-1"),
    value: HeaderValue.String("incredible"),
  },
];
const h2 = [
  {
    key: HeaderKeyFactory.String("x-header-bool"),
    value: HeaderValue.Bool(false),
  },
];

const messages = [
  { payload: "content with header", headers: h0 },
  { payload: "content solo" },
  { payload: "yolo msg" },
  { payload: "yolo msg 2" },
  { payload: "this is fuu", headers: h1 },
  { payload: "this is bar", headers: h2 },
  { payload: "yolo msg 3" },
  { payload: "fuu again", headers: h1 },
  { payload: "damnit", headers: h0 },
  { payload: "yolo msg 4", headers: h2 },
];

export const someMessageContent = () =>
  messages[Math.floor(Math.random() * messages.length)];

export const generateMessages = (count = 1) => {
  return [...Array(count)].map(() => ({ id: v7(), ...someMessageContent() }));
};

export const sendSomeMessages =
  (s: ClientProvider) =>
  async (streamId: Id, topicId: Id, partition: Partitioning) => {
    const rSend = await sendMessages(s)({
      topicId,
      streamId,
      messages: generateMessages(100),
      partition,
    });
    assert.ok(rSend);
    return rSend;
  };

export const formatPolledMessages = (msgs: Message[]) =>
  msgs.map((m) => {
    const {
      headers: { id, offset, timestamp, checksum },
      payload,
      userHeaders,
    } = m;
    return {
      id,
      offset,
      headers: userHeaders,
      payload: payload.toString(),
      timestamp,
      checksum,
    };
  });

export const getIggyAddress = (
  host = "127.0.0.1",
  port = 8090,
): [string, number] => {
  if (process.env.IGGY_TCP_ADDRESS) {
    const s = (process.env.IGGY_TCP_ADDRESS || "").split(":");
    [host, port] = [s[0], s[1] ? parseInt(s[1].toString(), 10) : port];
  }
  return [host, port];
};