fix: External rollups serverless implementation

cube-js · May 11, 2019 · 6d13370 · 6d13370
1 parent 0309d03
commit 6d13370
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 2 deletions.
diff --git a/packages/cubejs-query-orchestrator/orchestrator/QueryCache.js b/packages/cubejs-query-orchestrator/orchestrator/QueryCache.js
@@ -111,13 +111,13 @@ class QueryCache {
   getExternalQueue() {
     if (!this.externalQueue) {
       this.externalQueue = QueryCache.createQueue(
-        `SQL_QUERY_${this.redisPrefix}`,
+        `SQL_QUERY_EXT_${this.redisPrefix}`,
         this.externalDriverFactory,
         (client, q) => client.query(q.query, q.values),
         {
           logger: this.logger,
           concurrency: 6,
-          ...this.options.queueOptions
+          ...this.options.externalQueueOptions
         }
       );
     }

diff --git a/packages/cubejs-serverless/Handlers.js b/packages/cubejs-serverless/Handlers.js
@@ -12,6 +12,14 @@ const processHandlers = {
     const queue = orchestrator.queryCache.getQueue();
     await queue.processCancel(query);
   },
+  externalQueryProcess: async (queryKey, orchestrator) => {
+    const queue = orchestrator.queryCache.getExternalQueue();
+    await queue.processQuery(queryKey);
+  },
+  externalQueryCancel: async (query, orchestrator) => {
+    const queue = orchestrator.queryCache.getExternalQueue();
+    await queue.processCancel(query);
+  },
   preAggregationProcess: async (queryKey, orchestrator) => {
     const queue = orchestrator.preAggregations.getQueue();
     await queue.processQuery(queryKey);
@@ -30,6 +38,10 @@ class Handlers {
           queueOptions: {
             sendProcessMessageFn: async (queryKey) => this.sendNotificationMessage(queryKey, 'queryProcess', context),
             sendCancelMessageFn: async (query) => this.sendNotificationMessage(query, 'queryCancel', context)
+          },
+          externalQueueOptions: {
+            sendProcessMessageFn: async (queryKey) => this.sendNotificationMessage(queryKey, 'externalQueryProcess', context),
+            sendCancelMessageFn: async (query) => this.sendNotificationMessage(query, 'externalQueryCancel', context)
           }
         },
         preAggregationsOptions: {