Resolving merge conflicts

Author: MichaelCSnyder

.gitignore

@@ -109,6 +109,4 @@ package-lock.json
 local_ignore/
 .gitignore
 
-# KSQLDB docker server settings
-ksqldb_server_config/
 

.npmignore

@@ -0,0 +1,3 @@
+#Test Files
+/__tests__/
+.yml

README.md

@@ -45,8 +45,8 @@ npm install ksqldb-js
 Create a client in the application file:
 
 ```
-const ksqljs = require('ksqldb-js');
-const client = new ksqljs({ksqldbURL: '<url to ksqlDB server>'})
+const ksqldb = require('ksqldb-js');
+const client = new ksqldb({ksqldbURL: '<url to ksqlDB server>'})
 ```
 
 To run tests initiate Docker containers included in yaml file:

__tests__/integrationtests.js

@@ -1,5 +1,5 @@
 const { default: waitForExpect } = require('wait-for-expect');
-const ksqljs = require('../ksqljs/ksqlJS.js');
+const ksqldb = require('../ksqldb/ksqldb');
 // Pre-requisite: start a docker container
 /* To add to README: Prior to running test with 'npm test', please start the ksqlDB
 server using the command 'docker-compose up'. This will spin up a ksqlDB server on
@@ -16,7 +16,7 @@ need to be removed first.
 describe('--Integration Tests--', () => {
   describe('--Method Tests--', () => {
     beforeAll((done) => {
-      client = new ksqljs({ ksqldbURL: 'http://localhost:8088' });
+      client = new ksqldb({ ksqldbURL: 'http://localhost:8088' });
       done();
     });
 
@@ -25,6 +25,7 @@ describe('--Integration Tests--', () => {
     })
 
     it('.createStream properly creates a stream', async () => {
+      await client.ksql('DROP STREAM IF EXISTS TESTJESTSTREAM DELETE TOPIC;')
       const result = await client.createStream('TESTJESTSTREAM', ['name VARCHAR', 'email varchar', 'age INTEGER'], 'testJestTopic', 'json', 1);
       const streams = await client.ksql('LIST STREAMS;');
       const allStreams = streams.streams;
@@ -56,9 +57,7 @@ describe('--Integration Tests--', () => {
     })
 
     it('.insertStream properly inserts a row into a stream', async () => {
-      const response = await client.insertStream('TESTJESTSTREAM', [
-        { "name": "stab-rabbit", "email": "123@mail.com", "age": 100 }
-      ]);
+      
       const data = [];
       await client.push('SELECT * FROM TESTJESTSTREAM EMIT CHANGES;', async (chunk) => {
         data.push(JSON.parse(chunk));
@@ -67,6 +66,9 @@ describe('--Integration Tests--', () => {
           expect(data[1]).toEqual(["stab-rabbit", "123@mail.com", 100])
         }
       });
+      const response = await client.insertStream('TESTJESTSTREAM', [
+        { "name": "stab-rabbit", "email": "123@mail.com", "age": 100 }
+      ]);
     })
 
     it('.pull receives the correct data from a pull query', async () => {
@@ -98,24 +100,24 @@ describe('--Integration Tests--', () => {
         beforeAll(async () => {
           // await client.ksql('DROP STREAM IF EXISTS testAsStream;')
           // await client.ksql('DROP STREAM IF EXISTS newTestStream DELETE TOPIC;');
-  
+
           // await client.createStream('newTestStream', ['name VARCHAR', 'age INTEGER'], 'newTestTopic', 'json', 1);
           testAsQueryId = await client.createStreamAs('testAsStream', ['name', 'age'], 'newTestStream', {
             kafka_topic: 'newTestTopic',
             value_format: 'json',
             partitions: 1
-        }, 'age > 50');
+          }, 'age > 50');
         })
-  
+
         afterAll(async () => {
           await client.ksql('DROP STREAM IF EXISTS testAsStream;')
           // await client.ksql('DROP STREAM IF EXISTS newTestStream DELETE TOPIC;');
         })
-  
+
         it('creates materialized stream', async () => {
           let streamFound = false;
-          const {streams} = await client.ksql('LIST STREAMS;');
-  
+          const { streams } = await client.ksql('LIST STREAMS;');
+
           for (let i = 0; i < streams.length; i++) {
             if (streams[i].name, streams[i].name === 'TESTASSTREAM') {
               streamFound = true;
@@ -128,8 +130,8 @@ describe('--Integration Tests--', () => {
 
 
       describe('--Materialized Tables Tests--', () => {
-        beforeAll( async () => {
-          await client.createTableAs('testAsTable', 'newTestStream', ['name', 'LATEST_BY_OFFSET(age) AS recentAge'], {topic:'newTestTopic'},{WHERE: 'age >= 21', GROUP_BY: 'name'});
+        beforeAll(async () => {
+          await client.createTableAs('testAsTable', 'newTestStream', ['name', 'LATEST_BY_OFFSET(age) AS recentAge'], { topic: 'newTestTopic' }, { WHERE: 'age >= 21', GROUP_BY: 'name' });
         });
         afterAll(async () => {
           await client.ksql('DROP TABLE IF EXISTS testAsTable;');
@@ -138,37 +140,24 @@ describe('--Integration Tests--', () => {
         })
 
         it('creates a materialized table view of a stream', async () => {
-          const {tables} = await client.ksql('LIST TABLES;');
+          const { tables } = await client.ksql('LIST TABLES;');
           let tableFound = false;
-          for (let i = 0; i < tables.length; i++){
+          for (let i = 0; i < tables.length; i++) {
             if (tables[i].name === 'TESTASTABLE') {
               tableFound = true;
               break;
             }
           }
           expect(tableFound).toEqual(true);
         })
-
-        // it('receives updates from source stream', async () => {
-        //   let rowReceived = false;
-        //   await client.push('SELECT * FROM testAsTable EMIT CHANGES LIMIT 1;', async (data) => {
-        //     if (Array.isArray(JSON.parse(data))){
-        //       if (JSON.parse(data)[0] === "firstTester" && JSON.parse(data)[1] === 25){
-        //         rowReceived = true;
-        //       }
-        //     }
-        //   })
-        //   await client.insertStream('NEWTESTSTREAM', [{"NAME":"firstTester", "AGE":25}]);
-        //   await waitForExpect(() => expect(rowReceived).toEqual(true))
-        // })
       })
     })
   })
 
 
   describe('--Health Tests--', () => {
     beforeAll((done) => {
-      client = new ksqljs({ ksqldbURL: 'http://localhost:8088' });
+      client = new ksqldb({ ksqldbURL: 'http://localhost:8088' });
       done();
     });
 

__tests__/queryBuilderTests.js

@@ -1,5 +1,5 @@
-const queryBuilder = require('../ksqljs/queryBuilder.js');
-const { QueryBuilderError, EmptyQueryError, NumParamsError, InappropriateStringParamError } = require('../ksqljs/customErrors.js');
+const queryBuilder = require('../ksqldb/queryBuilder.js');
+const { QueryBuilderError, EmptyQueryError, NumParamsError, InappropriateStringParamError } = require('../ksqldb/customErrors.js');
 
 
 describe('Unit tests for query builder class', () => {

klip_64_ksqldb_js_client.md

@@ -6,7 +6,21 @@ const validateInputs = require('./validateInputs.js');
 const queryBuilder = require('./queryBuilder.js');
 const builder = new queryBuilder();
 
-class ksqljs {
+class ksqldb {
+  /**
+   * Constructor
+   * @param {object} config 
+   * 
+   * Config object can have these properties 
+   * 
+   * ksqldbURL: Connection URL or address
+   * 
+   * API: Username or API key for basic authentication
+   * 
+   * secret: Password or secret for basic authentication
+   * 
+   * httpsAgent: httpsAgent for setting TLS properties
+   */
   constructor(config) {
     this.ksqldbURL = config.ksqldbURL;
     this.API = config.API ? config.API : null;
@@ -274,7 +288,16 @@ class ksqljs {
     // utilize query with variables to build actual query
     const query = builder.build(builderQuery, [streamName], ...propertiesArgs, [sourceStream]);
 
-    return axios.post(this.ksqldbURL + '/ksql', { ksql: query })
+    return axios.post(this.ksqldbURL + '/ksql', { ksql: query }, {
+      headers:
+        this.API && this.secret ?
+          {
+            "Authorization": `Basic ${Buffer.from(this.API + ":" + this.secret, 'utf8').toString('base64')}`,
+          }
+          :
+          {},
+      httpsAgent: this.httpsAgentAxios ? this.httpsAgentAxios : null,
+    })
       .then(res => res.data[0].commandStatus.queryId)
       .catch(error => {
         throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
@@ -349,34 +372,34 @@ class ksqljs {
       partitions: 1
     };
     Object.assign(defaultProps, propertiesObj);
-    
+
     // if there's no properties Obj, assign them all default values
 
     // expect user to input a conditions object of format {WHERE: condition, GROUP_BY: condition, HAVING: condition};
     // generate conditions string based on object
     // const builder = new queryBuilder();
 
     let conditionQuery = '';
-    if (conditionsObj){
+    if (conditionsObj) {
       const conditionsArr = ['WHERE', 'GROUP_BY', 'HAVING'];
       const sqlClauses = [];
-    
+
       let i = 0;
-      while (conditionsArr.length){
-        if (conditionsObj[conditionsArr[0]]){
-          sqlClauses[i] = [conditionsArr[0].replace('_',' ')]; // clause values are set as arrays for query builder
-          sqlClauses[i+1] =[' ' + conditionsObj[conditionsArr[0]] + ' '];
+      while (conditionsArr.length) {
+        if (conditionsObj[conditionsArr[0]]) {
+          sqlClauses[i] = [conditionsArr[0].replace('_', ' ')]; // clause values are set as arrays for query builder
+          sqlClauses[i + 1] = [' ' + conditionsObj[conditionsArr[0]] + ' '];
         }
         else {
           sqlClauses[i] = [''];
-          sqlClauses[i+1] = [''];
+          sqlClauses[i + 1] = [''];
         }
-        i+=2;
+        i += 2;
         conditionsArr.shift()
       }
       conditionQuery = builder.build('??????', sqlClauses[0], sqlClauses[1], sqlClauses[2], sqlClauses[3], sqlClauses[4], sqlClauses[5]);
     }
-  
+
 
     // reformat for builder
     tableName = [tableName];
@@ -386,23 +409,30 @@ class ksqljs {
 
 
     const query = builder.build(`CREATE TABLE ? WITH (kafka_topic=?, value_format=?, partitions=?) AS SELECT ? FROM ? ?EMIT CHANGES;`, tableName, defaultProps.topic, defaultProps.value_format, defaultProps.partitions, selectColStr, source, conditionQuery)
-    return axios.post(this.ksqldbURL + '/ksql', { ksql: query })
-    .catch(error =>{
-      throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
-    });
+    return axios.post(this.ksqldbURL + '/ksql', { ksql: query }, {
+      headers:
+        this.API && this.secret ?
+          {
+            "Authorization": `Basic ${Buffer.from(this.API + ":" + this.secret, 'utf8').toString('base64')}`,
+          }
+          :
+          {},
+      httpsAgent: this.httpsAgentAxios ? this.httpsAgentAxios : null,
+    })
+      .catch(error => console.log(error));
   }
 
-    /**
-   * Inserts rows of data into a stream.
-   *
-   * <p>This method may be used to insert new rows of data into a stream.
-   *
-   * <p>This method is sql injection protected with the use of queryBuilder.
-   *
-   * @param {string} stream the name of the stream to insert data into.
-   * @param {object} rows an array that contains data that is being inserted into the stream.
-   * @return {Promise} this method returns a promise that resolves into an array describing the status of the row inserted.
-   */
+  /**
+ * Inserts rows of data into a stream.
+ *
+ * <p>This method may be used to insert new rows of data into a stream.
+ *
+ * <p>This method is sql injection protected with the use of queryBuilder.
+ *
+ * @param {string} stream the name of the stream to insert data into.
+ * @param {object} rows an array that contains data that is being inserted into the stream.
+ * @return {Promise} this method returns a promise that resolves into an array describing the status of the row inserted.
+ */
   //---------------------Insert Rows Into Existing Streams-----------------
   insertStream = (stream, rows) => {
     validateInputs([stream, 'string', 'stream', true], [rows, 'array', 'rows', true]);
@@ -624,4 +654,4 @@ class ksqljs {
   }
 };
 
-module.exports = ksqljs;
+module.exports = ksqldb;