Merge pull request #21 from oslabs-beta/ms/createStreamAs

Ms/create stream as

Author: javanang

__tests__/integrationtests.js

@@ -53,7 +53,6 @@ describe('--Integration Tests--', () => {
       const response = await client.insertStream('TESTJESTSTREAM', [
         { "name": "stab-rabbit", "email": "123@mail.com", "age": 100 }
       ]);
-      console.log(response);
       const data = [];
       await client.push('SELECT * FROM TESTJESTSTREAM EMIT CHANGES;', async (chunk) => {
         data.push(JSON.parse(chunk));

ksqljs/ksqlJS.js

@@ -155,7 +155,6 @@ class ksqljs {
    * @return {Promise} a promise that completes once the server response is received, and returns a response object.
    */
   createStream(name, columnsType, topic, value_format = 'json', partitions = 1, key) {
-    // console.log(this.ksqldbURL);
     if (typeof name !== 'string' || typeof columnsType !== 'object' || typeof topic !== 'string' || typeof partitions !== 'number') {
       return console.log("invalid input(s)")
     }
@@ -170,6 +169,50 @@ class ksqljs {
       });
   }
 
+  /**
+   * 
+   * @param {string} streamName 
+   * @param {string[]} selectColumns 
+   * @param {string} sourceStream 
+   * @param {object} propertiesObj 
+   * @param {string} conditions 
+   * @param {string} partitionBy 
+   * @returns {Promise}
+   */
+  createStreamAs = (streamName, selectColumns, sourceStream, propertiesObj, conditions, partitionBy) => {
+    const propertiesArgs = [];
+    const selectColStr = selectColumns.reduce((result, current) => result + ', ' + current);
+    // begin with first consistent portion of query
+    let builderQuery = 'CREATE STREAM ? ';
+    
+    // include properties in query if provided
+    if(Object.keys(propertiesObj).length > 0) {
+      builderQuery += 'WITH (';
+      for (const [key, value] of Object.entries(propertiesObj)) {
+        const justStarted = builderQuery[builderQuery.length - 1] === '(';
+
+        if (!justStarted) builderQuery += ', ';
+        builderQuery += '? = ?';
+        propertiesArgs.push([key], value);
+      };
+      builderQuery += ') ';
+    }
+
+    // continue building the query to be sent to the builder
+    builderQuery += `AS SELECT ${selectColStr} FROM ? `;
+    if (conditions.indexOf(';') === -1) builderQuery += `WHERE ${conditions} `;
+    builderQuery += partitionBy || '';
+    builderQuery += 'EMIT CHANGES;'
+
+    // utilize query with variables to build actual query
+    const query = builder.build(builderQuery, [streamName], ...propertiesArgs, [sourceStream]);
+
+    return axios.post(this.ksqldbURL + '/ksql', { ksql: query })
+    .then(res => res.data[0].commandStatus.queryId)
+    .catch(error => console.log(error));
+  }
+
+  //---------------------Create tables-----------------
   /**
    * Executes a query to create a table.
    *

ksqljsTest.js

@@ -17,41 +17,19 @@ let metadata;
     console.log('this is the result', result);
 }
 
-pullTest();
+pullTest(); */
 
 //---------------------Test Push Queries-------------------
-const pushTest = async () => {
-    // metadata = await client.push('SELECT * FROM riderlocations EMIT CHANGES LIMIT 1;', (row) => console.log(row));
-    // console.log('this is the metadata returned ', metadata);
-    let pushActive = false;
-    await client.createStream('TESTJESTSTREAM', ['age INTEGER'], 'testJestTopic', 'json', 1);
-    // await client.push('SELECT * FROM TESTJESTSTREAM EMIT CHANGES LIMIT 1;', (data) => {
-    //   console.log(data);
-    //   console.log('HERE IS DATA ', JSON.parse(data).queryId)
-    //   if(JSON.parse(data).queryId){
-    //     pushActive = true;
-    //   }
-    //   client.ksql(`TERMINATE ${JSON.parse(data).queryId};`)
-    //   client.ksql('DROP STREAM IF EXISTS TESTJESTSTREAM DELETE TOPIC;');
-    // });
-    // await client.ksql('DROP STREAM IF EXISTS TESTJESTSTREAM DELETE TOPIC;');
-
-    /* const pushTest = async () => {
+/* const pushTest = async () => {
     try {
         metadata = await client.push('SELECT * FROM riderlocations EMIT CHANGES LIMIT 1;', (row) => console.log(row));
         console.log('this is the metadata returned ', metadata);
     } catch (error) {
         console.log(error);
     }
-
-    metadata = await client.push('SELECT * FROM riderlocations EMIT CHANGES LIMIT 1;', (row) => console.log(row));
-    console.log('this is the metadata returned ', metadata);
-
-    metadata = await client.push('SELECT * FROM riderlocations EMIT CHANGES LIMIT 1;', (row) => console.log(row));
-    console.log('this is the metadata returned ', metadata);
 };
 
-pushTest();
+pushTest(); */
 
 //---------------------Test Termination of Queries-------------------
 /* const terminateTest = async () => {
@@ -71,11 +49,11 @@ setTimeout(() => terminateTest(metadata), 2000); */
 listQueries(); */
 
 //---------------------Test Stream Creation-------------------
-// const createStreamTest = () => {
-//     client.createStream('TestStream', ['name VARCHAR', 'email varchar', 'age INTEGER'], 'testTopic', 'json', 1);
-// }
+/* const createStreamTest = () => {
+    client.createStream('TestStream', ['name VARCHAR', 'email varchar', 'age INTEGER'], 'testTopic', 'json', 1);
+}
 
-// createStreamTest();
+createStreamTest(); */
 
 //---------------------Test Table Creation-------------------
 /* const createTableTest = () => {
@@ -85,45 +63,42 @@ listQueries(); */
 createTableTest(); */
 
 //---------------------Test Insert Stream-------------------
-/* const insertStreamTest = () => {
-    client.insertStream('TestStream', [
-        { "name": "matt", "email": "123@mail.com", "age": 1000 },
-        { "name": "jonathan", "email": "234@mail.com", "age": 99 }
-    ]);
-/* const insertStreamTest = async () => {
+/*   const insertStreamTest = async () => {
     const test = await client.insertStream('TestStream', [
         { "name": "matt", "email": "123@mail.com", "age": 1000 },
         { "name": "jonathan", "email": "234@mail.com", "age": 99 }
     ]);
     console.log('returned array: ', test);
-*/
-
-const insertStreamTest = async () => {
-    // const test = await client.insertStream('TestStream', [
-        // { "name": "Scrooge", "email": "mcduck@mail.com", "age": 59 },
-        // { "name": "jonathan", "email": "234@mail.com", "age": 99 }
-    // ]);
-    // console.log('returned array: ', test);
-};
-
-// insertStreamTest();
+  };
+  insertStreamTest(); */
 
-const pullFromToTest = async () => {
+/* const pullFromToTest = async () => {
     const data = await client.pullFromTo('TESTSTREAM', 'America/Los_Angeles', ['2022-05-18', '00', '00', '00'], ['2022-05-20', '00', '00', '00']);
     // console.log(data);
 }
 
-pullFromTo();
-// const insertStreamTest = async () => {
-//     const test = await client.insertStream('TestStream', [
-//         { "name": "matt", "email": "123@mail.com", "age": 1000 },
-//         { "name": "jonathan", "email": "234@mail.com", "age": 99 }
-//     ]);
-//     // console.log('returned array: ', test);
-// };
+pullFromTo(); */
+
+//---------------------Test Stream Creation As-------------------
+
+const createStreamAsTest = async () => {
+    const queryId = await client.createStreamAs('TestAsStream', ['latitude', 'longitude'], 'riderLocations', {
+        kafka_topic: 'TestAsStream',
+        value_format: 'json',
+        partitions: 1
+    },
+    'latitude > 37');
+    console.log('this is the queryId: ', queryId);
+};
 
-// insertStreamTest();
+createStreamAsTest();
 
+// (streamName, selectColumns, sourceStream, propertiesObj, conditionsObj)
+//   const defaultProps = {
+//       kafka_topic: streamName,
+//       value_format: 'json',
+//       partitions: 1
+//     };
 
 //---------------------Test Inspect query status -------------------
 // const inspectQueryStatusTest = async () => {
@@ -208,4 +183,4 @@ pullFromTo();
 //     // should return true
 // };
 
-// isValidPropertyTest();
+// isValidPropertyTest();