make parse_cubestore_ws_result_message async

Author: KSDaemon

packages/cubejs-backend-native/js/index.ts

@@ -370,10 +370,10 @@ export const buildSqlAndParams = (cubeEvaluator: any): String => {
 
 export type ResultRow = Record<string, string>;
 
-export const parseCubestoreResultMessage = (message: ArrayBuffer): CubeStoreResultWrapper => {
+export const parseCubestoreResultMessage = async (message: ArrayBuffer): Promise<CubeStoreResultWrapper> => {
   const native = loadNative();
 
-  const msg = native.parseCubestoreResultMessage(message);
+  const msg = await native.parseCubestoreResultMessage(message);
   return new CubeStoreResultWrapper(msg);
 };
 

packages/cubejs-backend-native/src/node_export.rs

@@ -515,13 +515,20 @@ fn debug_js_to_clrepr_to_js(mut cx: FunctionContext) -> JsResult<JsValue> {
 
 //============ sql orchestrator ===================
 
-fn parse_cubestore_ws_result_message(mut cx: FunctionContext) -> JsResult<JsBox<CubeStoreResult>> {
+fn parse_cubestore_ws_result_message(mut cx: FunctionContext) -> JsResult<JsPromise> {
     let msg = cx.argument::<JsBuffer>(0)?;
-    let msg_data = msg.as_slice(&cx);
-    match CubeStoreResult::new(msg_data) {
-        Ok(result) => Ok(cx.boxed(result)),
-        Err(err) => cx.throw_error(err.to_string()),
-    }
+    let msg_data = msg.as_slice(&cx).to_vec();
+
+    let promise = cx
+        .task(move || CubeStoreResult::new(&msg_data))
+        .promise(move |mut cx, res| {
+            match res {
+                Ok(result) => Ok(cx.boxed(result)),
+                Err(err) => cx.throw_error(err.to_string()),
+            }
+        });
+
+    Ok(promise)
 }
 
 fn get_cubestore_result(mut cx: FunctionContext) -> JsResult<JsValue> {

packages/cubejs-cubestore-driver/src/WebSocketConnection.ts

@@ -109,7 +109,7 @@ export class WebSocketConnection {
             this.webSocket = undefined;
           }
         });
-        webSocket.on('message', (msg) => {
+        webSocket.on('message', async (msg) => {
           const buf = new flatbuffers.ByteBuffer(msg);
           const httpMessage = HttpMessage.getRootAsHttpMessage(buf);
           const resolvers = webSocket.sentMessages[httpMessage.messageId()];
@@ -120,7 +120,7 @@ export class WebSocketConnection {
 
           if (getEnv('nativeOrchestrator') && msg.length > 1000) {
             try {
-              const nativeResMsg = parseCubestoreResultMessage(msg);
+              const nativeResMsg = await parseCubestoreResultMessage(msg);
               resolvers.resolve(nativeResMsg);
             } catch (e) {
               resolvers.reject(e);