Initial Commit that contains about half of the client functionality

Author: MichaelCSnyder

docker-compose.yml

@@ -0,0 +1,54 @@
+---
+version: "2"
+
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.0.1
+    hostname: zookeeper
+    container_name: zookeeper
+    ports:
+      - "2181:2181"
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+
+  broker:
+    image: confluentinc/cp-kafka:7.0.1
+    hostname: broker
+    container_name: broker
+    depends_on:
+      - zookeeper
+    ports:
+      - "29092:29092"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9092,PLAINTEXT_HOST://localhost:29092
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+
+  ksqldb-server:
+    image: confluentinc/ksqldb-server:0.25.1
+    hostname: ksqldb-server
+    container_name: ksqldb-server
+    depends_on:
+      - broker
+    ports:
+      - "8088:8088"
+    environment:
+      KSQL_LISTENERS: http://0.0.0.0:8088
+      KSQL_BOOTSTRAP_SERVERS: broker:9092
+      KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: "true"
+      KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: "true"
+
+  ksqldb-cli:
+    image: confluentinc/ksqldb-cli:0.25.1
+    container_name: ksqldb-cli
+    depends_on:
+      - broker
+      - ksqldb-server
+    entrypoint: /bin/sh
+    tty: true

ksqljs/ksqlJS.js

@@ -0,0 +1,79 @@
+const axios = require("axios");
+const http2 = require("http2");
+
+const ksqljs = (ksqldbURL) => {
+  return {
+    //---------------------Pull queries (fetch a single batch of existing rows)-----------------
+    pull: (query) => {
+      return axios
+        .post(ksqldbURL + "/query-stream", {
+          sql: query,
+        })
+        .then((res) => res.data)
+        .catch((error) => console.log(error));
+    },
+    //---------------------Push queries (continue to receive updates to stream-----------------
+    push: (query, cb) => {
+      let sentQueryId = false;
+      let queryMetadata;
+      const session = http2.connect(ksqldbURL);
+
+      session.on("error", (err) => console.error(err));
+
+      const req = session.request({
+        ":path": "/query-stream",
+        ":method": "POST",
+      });
+
+      const reqBody = {
+        sql: query,
+        Accept: "application/json, application/vnd.ksqlapi.delimited.v1",
+      };
+
+      req.write(JSON.stringify(reqBody), "utf8");
+      req.end();
+      req.setEncoding("utf8");
+
+      req.on("data", (chunk) => {
+        if (!sentQueryId) {
+          sentQueryId = true;
+          queryMetadata = chunk;
+        }
+        cb(chunk);
+      });
+
+      req.on("end", () => session.close());
+
+        return new Promise((resolve, reject) => {
+            setTimeout(() => resolve(JSON.parse(queryMetadata)?.queryId), 1000);
+        })
+    },
+    //---------------------Terminate existing push queries-----------------
+    terminate: (queryId) => {
+        return axios.post(ksqldbURL + '/close-query', {queryId: queryId})
+        .then(res => res)
+        .catch(error => console.log(error));
+    },
+    //---------------------List existing streams, tables, topics, and queries-----------------
+    ksql: (query) => {
+        return axios.post(ksqldbURL + '/ksql', {ksql: query})
+        .then(res => res.data[0])
+        .catch(error => console.log(error));
+    },
+    //---------------------Create streams-----------------
+    createStream: (name, columnsType, topic, value_format = 'json', partitions=1, key) => {
+        const columnsTypeString = columnsType.reduce((result, currentType) => result + ', ' + currentType);
+        const query = `CREATE STREAM ${name} (${columnsTypeString}) WITH (kafka_topic='${topic}', value_format='${value_format}', partitions=${partitions});`;
+
+        axios.post(ksqldbURL + '/ksql', {ksql: query})
+        .catch(error => console.log(error));
+    },
+    //---------------------Create tables-----------------
+    createTable: (name, columnsType, topic, value_format = 'json') => {
+
+    }
+  };
+};
+
+module.exports = ksqljs;
+ 

ksqljsTest.js

@@ -0,0 +1,49 @@
+const ksqljs = require('./ksqlJS/ksqlJS.js');
+
+const client = ksqljs('http://localhost:8088');
+let metadata;
+
+//---------------------Test PUll Queries-------------------
+/* const pullTest = async () => {
+    const result = await client.pull('SELECT * FROM usersStream;');
+    console.log('this is the result', result);
+}
+
+pullTest(); */
+
+//---------------------Test Push Queries-------------------
+/* const pushTest = async () => {
+    metadata = await client.push('SELECT * FROM usersStream EMIT CHANGES LIMIT 1;', (row) => console.log(row));
+    console.log('this is the metadata returned ', metadata);
+};
+
+pushTest(); */
+
+//---------------------Test Termination of Queries-------------------
+/* const terminateTest = async () => {
+    client.terminate(metadata);
+};
+
+setTimeout(() => terminateTest(metadata), 2000); */
+
+//---------------------Test List Queries-------------------
+/* const listQueries = async () => {
+    console.log(await client.ksql('LIST QUERIES;'));
+    console.log(await client.ksql('LIST STREAMS;'));
+    console.log(await client.ksql('LIST TABLES;'));
+    console.log(await client.ksql('LIST TOPICS;'));
+}
+
+listQueries(); */
+
+//---------------------Test Stream Creation-------------------
+/* const createStreamTest = () => {
+    client.createStream('AnotherTestStream', ['name VARCHAR','email varchar','age INTEGER'], 'testTopic', 'json', 1);
+}
+
+createStreamTest(); */
+
+//---------------------Test Table Creation-------------------
+/* const createTableTest = () => {
+    client.createTable();
+} */