feat: added stream query for real time data updates

lowcoder-org · FalkWolsky · Sep 14, 2023 · Aug 31, 2023 · Sep 10, 2023 · Sep 13, 2023
commit 114fb380cc506a25f20bf97681e71fe8915e8d96
diff --git a/client/packages/lowcoder/src/components/ResCreatePanel.tsx b/client/packages/lowcoder/src/components/ResCreatePanel.tsx
@@ -161,6 +161,13 @@ const ResButton = (props: {
         dataSourceId: QUICK_REST_API_ID,
       },
     },
+    streamApi: {
+      label: trans("query.quickStreamAPI"),
+      type: BottomResTypeEnum.Query,
+      extra: {
+        compType: "streamApi",
+      },
+    },
     graphql: {
       label: trans("query.quickGraphql"),
       type: BottomResTypeEnum.Query,
@@ -318,6 +325,7 @@ export function ResCreatePanel(props: ResCreateModalProps) {
             <div className="section">
               <DataSourceListWrapper placement={placement}>
                 <ResButton size={buttonSize} identifier={"restApi"} onSelect={onSelect} />
+                <ResButton size={buttonSize} identifier={"streamApi"} onSelect={onSelect} />
                 <ResButton size={buttonSize} identifier={"graphql"} onSelect={onSelect} />
                 {placement === "editor" && (
                   <ResButton size={buttonSize} identifier={"lowcoderApi"} onSelect={onSelect} />

diff --git a/client/packages/lowcoder/src/comps/queries/httpQuery/httpQueryConstants.tsx b/client/packages/lowcoder/src/comps/queries/httpQuery/httpQueryConstants.tsx
@@ -5,6 +5,7 @@ import { HttpQuery } from "./httpQuery";
 import styled from "styled-components";
 import { QueryConfigItemWrapper, QueryConfigLabel, QueryConfigWrapper } from "components/query";
 import { GraphqlQuery } from "./graphqlQuery";
+import { StreamQuery } from "./streamQuery";
 
 const UrlInput = styled.div<{ hasAddonBefore: boolean }>`
   display: flex;
@@ -33,7 +34,7 @@ const UrlInputAddonBefore = styled.div`
 `;
 
 export const HttpPathPropertyView = (props: {
-  comp: InstanceType<typeof HttpQuery | typeof GraphqlQuery>;
+  comp: InstanceType<typeof HttpQuery | typeof GraphqlQuery | typeof StreamQuery>;
   datasourceId: string;
   urlPlaceholder?: string;
 }) => {

diff --git a/client/packages/lowcoder/src/comps/queries/httpQuery/streamQuery.tsx b/client/packages/lowcoder/src/comps/queries/httpQuery/streamQuery.tsx
@@ -0,0 +1,276 @@
+import { ControlPropertyViewWrapper } from "components/control";
+import { Input } from "components/Input";
+import { KeyValueList } from "components/keyValueList";
+import { QueryConfigItemWrapper, QueryConfigLabel, QueryConfigWrapper } from "components/query";
+import { simpleMultiComp } from "comps/generators/multi";
+import { ReactNode } from "react";
+import { JSONValue } from "../../../util/jsonTypes";
+import { keyValueListControl } from "../../controls/keyValueControl";
+import { ParamsJsonControl, ParamsStringControl } from "../../controls/paramsControl";
+import { list } from "../../generators/list";
+import { valueComp, withDefault } from "../../generators/simpleGenerators";
+import { FunctionProperty, toQueryView } from "../queryCompUtils";
+import {
+  HttpHeaderPropertyView,
+  HttpParametersPropertyView,
+  HttpPathPropertyView,
+} from "./httpQueryConstants";
+import { QueryResult } from "../queryComp";
+import { QUERY_EXECUTION_ERROR, QUERY_EXECUTION_OK } from "constants/queryConstants";
+import { FunctionControl } from "comps/controls/codeControl";
+
+const connect = async (socket: WebSocket, timeout = 10000) => {
+  const isOpened = () => (socket.readyState === WebSocket.OPEN)
+
+  if (socket.readyState !== WebSocket.CONNECTING) {
+    return isOpened()
+  }
+  else {
+    const intrasleep = 100
+    const ttl = timeout / intrasleep // time to loop
+    let loop = 0
+    while (socket.readyState === WebSocket.CONNECTING && loop < ttl) {
+      await new Promise(resolve => setTimeout(resolve, intrasleep))
+      loop++
+    }
+    return isOpened()
+  }
+}
+
+const childrenMap = {
+  path: ParamsStringControl,
+  destroySocketConnection: FunctionControl,
+};
+
+const StreamTmpQuery = simpleMultiComp(childrenMap);
+
+export class StreamQuery extends StreamTmpQuery {
+  private socket: WebSocket | undefined;
+
+  override getView() {
+    return async (
+      p: {
+        args?: Record<string, unknown>,
+        callback?: (result: QueryResult) => void
+      }
+    ): Promise<QueryResult> => {
+      const children = this.children;
+
+      try {
+        const timer = performance.now();
+        this.socket = new WebSocket(children.path.children.text.getView());
+
+        this.socket.onopen = function(e) {
+          console.log("[open] Connection established");
+        };
+
+        this.socket.onmessage = function(event) {
+          console.log(`[message] Data received from server: ${event.data}`);
+          if(typeof JSON.parse(event.data) === 'object') {
+            const result = {
+              data: JSON.parse(event.data),
+              code: QUERY_EXECUTION_OK,
+              success: true,
+              runTime: Number((performance.now() - timer).toFixed()),
+            }
+            p?.callback?.(result);
+          }
+        };
+
+        this.socket.onclose = function(event) {
+          if (event.wasClean) {
+            console.log(`[close] Connection closed cleanly, code=${event.code} reason=${event.reason}`);
+          } else {
+            console.log('[close] Connection died');
+          }
+        };
+
+        this.socket.onerror = function(error) {
+          throw new Error(error as any)
+        };
+
+        const isConnectionOpen = await connect(this.socket);
+        if(!isConnectionOpen) {
+          return {
+            success: false,
+            data: "",
+            code: QUERY_EXECUTION_ERROR,
+            message: "Socket connection failed",
+          };
+        }
+
+        return {
+          data: "",
+          code: QUERY_EXECUTION_OK,
+          success: true,
+          runTime: Number((performance.now() - timer).toFixed()),
+        };
+      } catch (e) {
+        return {
+          success: false,
+          data: "",
+          code: QUERY_EXECUTION_ERROR,
+          message: (e as any).message || "",
+        };
+      }
+    };
+  }
+
+  propertyView(props: { datasourceId: string }) {
+    return <PropertyView {...props} comp={this} />;
+  }
+
+  destroy() {
+    this.socket?.close();
+  }
+}
+
+const PropertyView = (props: { comp: InstanceType<typeof StreamQuery>; datasourceId: string }) => {
+  const { comp } = props;
+
+  return (
+    <>
+      <HttpPathPropertyView
+        {...props}
+        comp={comp}
+        urlPlaceholder="wss://www.example.com/socketserver"
+      />
+    </>
+  );
+};
+
+
+
+// import { ParamsStringControl } from "comps/controls/paramsControl";
+// import { FunctionControl, StringControl, codeControl } from "comps/controls/codeControl";
+// import { MultiCompBuilder } from "comps/generators";
+// import { QueryResult } from "../queryComp";
+// import { QueryTutorials } from "util/tutorialUtils";
+// import { DocLink } from "lowcoder-design";
+// import { getGlobalSettings } from "comps/utils/globalSettings";
+// import { trans } from "i18n";
+// import { QUERY_EXECUTION_ERROR, QUERY_EXECUTION_OK } from "constants/queryConstants";
+
+// const connect = async (socket: WebSocket, timeout = 10000) => {
+//   const isOpened = () => (socket.readyState === WebSocket.OPEN)
+
+//   if (socket.readyState !== WebSocket.CONNECTING) {
+//     return isOpened()
+//   }
+//   else {
+//     const intrasleep = 100
+//     const ttl = timeout / intrasleep // time to loop
+//     let loop = 0
+//     while (socket.readyState === WebSocket.CONNECTING && loop < ttl) {
+//       await new Promise(resolve => setTimeout(resolve, intrasleep))
+//       loop++
+//     }
+//     return isOpened()
+//   }
+// }
+
+// export const StreamQuery = (function () {
+//   const childrenMap = {
+//     path: StringControl,
+//     destroySocketConnection: FunctionControl,
+//   };
+//   return new MultiCompBuilder(childrenMap, (props) => {
+//     const { orgCommonSettings } = getGlobalSettings();
+//     const runInHost = !!orgCommonSettings?.runJavaScriptInHost;
+
+//     console.log(props.path);
+//     return async (
+//       p: {
+//         args?: Record<string, unknown>,
+//         callback?: (result: QueryResult) => void
+//       }
+//     ): Promise<QueryResult> => {
+//       console.log('Stream Query', props)
+
+//       try {
+//         const timer = performance.now();
+//         // const url = 'wss://free.blr2.piesocket.com/v3/1?api_key=yWUvGQggacrrTdXYjvTpRD5qhm4RIsglS7YJlKzp&notify_self=1'
+//         const socket = new WebSocket(props.path);
+
+//         props.destroySocketConnection = () => {
+//           socket.close();
+//         };
+
+//         socket.onopen = function(e) {
+//           console.log("[open] Connection established");
+//         };
+
+//         socket.onmessage = function(event) {
+//           console.log(`[message] Data received from server: ${event.data}`);
+//           console.log(JSON.parse(event.data))
+//           if(typeof JSON.parse(event.data) === 'object') {
+//             const result = {
+//               data: JSON.parse(event.data),
+//               code: QUERY_EXECUTION_OK,
+//               success: true,
+//               runTime: Number((performance.now() - timer).toFixed()),
+//             }
+//             p?.callback?.(result);
+//           }
+//         };
+
+//         socket.onclose = function(event) {
+//           if (event.wasClean) {
+//             console.log(`[close] Connection closed cleanly, code=${event.code} reason=${event.reason}`);
+//           } else {
+//             // e.g. server process killed or network down
+//             // event.code is usually 1006 in this case
+//             console.log('[close] Connection died');
+//           }
+//         };
+
+//         socket.onerror = function(error) {
+//           throw new Error(error as any)
+//         };
+//         const isConnectionOpen = await connect(socket);
+//         if(!isConnectionOpen) {
+//           return {
+//             success: false,
+//             data: "",
+//             code: QUERY_EXECUTION_ERROR,
+//             message: "Socket connection failed",
+//           };
+//         }
+
+//         // const data = await props.script(p.args, runInHost);
+//         return {
+//           data: "",
+//           code: QUERY_EXECUTION_OK,
+//           success: true,
+//           runTime: Number((performance.now() - timer).toFixed()),
+//         };
+//       } catch (e) {
+//         return {
+//           success: false,
+//           data: "",
+//           code: QUERY_EXECUTION_ERROR,
+//           message: (e as any).message || "",
+//         };
+//       }
+//     };
+//   })
+//     .setPropertyViewFn((children) => {
+//       return (
+//         <>
+//           {
+//             children.path.propertyView({
+//               label: "URL",
+//               placement: "bottom",
+//               placeholder:"wss://www.example.com/socketserver",
+//             })
+//           }
+
+//           {/* TODO: Add docs for Stream Query
+//           {QueryTutorials.js && (
+//             <DocLink href={QueryTutorials.js}>{trans("query.jsQueryDocLink")}</DocLink>
+//           )} */}
+//         </>
+//       );
+//     })
+//     .build();
+// })();
diff --git a/client/packages/lowcoder/src/comps/queries/queryComp.tsx b/client/packages/lowcoder/src/comps/queries/queryComp.tsx
@@ -69,6 +69,7 @@ import { millisecondsControl } from "../controls/millisecondControl";
 import { paramsMillisecondsControl } from "../controls/paramsControl";
 import { NameConfig, withExposingConfigs } from "../generators/withExposing";
 import { HttpQuery } from "./httpQuery/httpQuery";
+import { StreamQuery } from "./httpQuery/streamQuery";
 import { QueryConfirmationModal } from "./queryComp/queryConfirmationModal";
 import { QueryNotificationControl } from "./queryComp/queryNotificationControl";
 import { QueryPropertyView } from "./queryComp/queryPropertyView";
@@ -419,6 +420,7 @@ QueryCompTmp = class extends QueryCompTmp {
           applicationPath: parentApplicationPath,
           args: action.args,
           timeout: this.children.timeout,
+          callback: (result) => this.processResult(result, action, startTime)
         });
       }, getTriggerType(this) === "manual")
       .then(
@@ -517,6 +519,7 @@ QueryCompTmp = class extends QueryCompTmp implements BottomResComp {
     switch (type) {
       case "js":
       case "restApi":
+      case "streamApi":
       case "mongodb":
       case "redis":
       case "es":
@@ -708,6 +711,9 @@ class QueryListComp extends QueryListTmpComp implements BottomResListComp {
         ],
       })
     );
+    if(toDelQuery.children.compType.getView() === 'streamApi') {
+      (toDelQuery.children.comp as StreamQuery)?.destroy();
+    }
     messageInstance.success(trans("query.deleteSuccessMessage", { undoKey }));
   }
 

diff --git a/client/packages/lowcoder/src/comps/queries/queryComp/queryPropertyView.tsx b/client/packages/lowcoder/src/comps/queries/queryComp/queryPropertyView.tsx
@@ -307,7 +307,7 @@ export const QueryGeneralPropertyView = (props: {
                 [
                   {
                     label:
-                      children.compType.getView() === "js"
+                      (children.compType.getView() === "js" || children.compType.getView() === "streamApi")
                         ? trans("query.triggerTypePageLoad")
                         : trans("query.triggerTypeAuto"),
                     value: "automatic",
@@ -363,6 +363,7 @@ function useDatasourceStatus(datasourceId: string, datasourceType: ResourceType)
   return useMemo(() => {
     if (
       datasourceType === "js" ||
+      datasourceType === "streamApi" ||
       datasourceType === "libraryQuery" ||
       datasourceId === QUICK_REST_API_ID ||
       datasourceId === QUICK_GRAPHQL_ID ||

diff --git a/client/packages/lowcoder/src/comps/queries/resourceDropdown.tsx b/client/packages/lowcoder/src/comps/queries/resourceDropdown.tsx
@@ -91,6 +91,11 @@ const QuickRestAPIValue: ResourceOptionValue = {
   type: "restApi",
 };
 
+const QuickStreamAPIValue: ResourceOptionValue = {
+  id: "",
+  type: "streamApi",
+};
+
 const QuickGraphqlValue: ResourceOptionValue = {
   id: QUICK_GRAPHQL_ID,
   type: "graphql",
@@ -254,6 +259,17 @@ export const ResourceDropdown = (props: ResourceDropdownProps) => {
           </SelectOptionContains>
         </SelectOption>
 
+        <SelectOption
+          key={JSON.stringify(QuickStreamAPIValue)}
+          label={trans("query.quickStreamAPI")}
+          value={JSON.stringify(QuickStreamAPIValue)}
+        >
+          <SelectOptionContains>
+            {getBottomResIcon("restApi")}
+            <SelectOptionLabel>{trans("query.quickStreamAPI")} </SelectOptionLabel>
+          </SelectOptionContains>
+        </SelectOption>
+
         <SelectOption
           key={JSON.stringify(QuickGraphqlValue)}
           label={trans("query.quickGraphql")}