Created
April 8, 2020 18:53
-
-
Save chikka/3c66ea80b7082f3414602af6ed2a43ef to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
const aws = require('aws-sdk'); | |
const sns = new aws.SNS(); | |
const express = require('serverless-express/express'); | |
const handler = require('serverless-express/handler'); | |
const cors = require('cors'); | |
const ServerCore = require('@cubejs-backend/server-core'); | |
const secretManager = require('./awsSecretService.js') | |
const processHandlers = { | |
queryProcess: async (queryKey, orchestrator) => { | |
const queue = orchestrator.queryCache.getQueue(); | |
await queue.processQuery(queryKey); | |
}, | |
queryCancel: async (query, orchestrator) => { | |
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); | |
}, | |
preAggregationCancel: async (query, orchestrator) => { | |
const queue = orchestrator.preAggregations.getQueue(); | |
await queue.processCancel(query); | |
} | |
}; | |
class AWSHandlers { | |
constructor(options) { | |
options = { | |
orchestratorOptions: (context) => ({ | |
queryCacheOptions: { | |
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: { | |
queueOptions: { | |
sendProcessMessageFn: async (queryKey) => this.sendNotificationMessage(queryKey, 'preAggregationProcess', context), | |
sendCancelMessageFn: async (query) => this.sendNotificationMessage(query, 'preAggregationCancel', context) | |
} | |
} | |
}), | |
...options | |
}; | |
this.serverCore = new ServerCore(options); | |
this.api = this.api.bind(this); | |
this.process = this.process.bind(this); | |
} | |
topicArn(topic) { | |
return `arn:aws:sns:${process.env.AWS_REGION}:${process.env.AWS_ACCOUNT_ID}:${topic}`; | |
} | |
async sendNotificationMessage(message, type, context) { | |
const params = { | |
Message: JSON.stringify({ message, type, context }), | |
TopicArn: this.topicArn(`${process.env.CUBEJS_APP || 'cubejs'}-process`) | |
}; | |
await sns.publish(params).promise(); | |
} | |
async process(event) { | |
await Promise.all(event.Records.map(async record => { | |
const message = JSON.parse(record.Sns.Message); | |
await this.processMessage(message); | |
})); | |
return { | |
statusCode: 200 | |
}; | |
} | |
async getApiHandler() { | |
if (!this.apiHandler) { | |
console.log("getApiHandler") | |
const secret = await secretManager.getSecretValue("rds-db-credentials-kp-test") | |
console.log('Hello!') | |
process.env.CUBEJS_DB_HOST = secret.host | |
process.env.CUBEJS_DB_NAME = "df_analytics" | |
process.env.CUBEJS_DB_USER = secret.username | |
process.env.CUBEJS_DB_PASS = secret.password | |
process.env.CUBEJS_DB_PORT = secret.port | |
const app = express(); | |
app.use(cors()); | |
this.serverCore.initApp(app); | |
this.apiHandler = handler(app); | |
} | |
return this.apiHandler; | |
} | |
async api(event, context) { | |
const data = await this.getApiHandler() | |
return await data(event, context); | |
} | |
async processMessage(message) { | |
const processFn = processHandlers[message.type]; | |
if (!processFn) { | |
throw new Error(`Unrecognized message type: ${message.type}`); | |
} | |
const orchestratorApi = this.serverCore.getOrchestratorApi(message.context); | |
await processFn(message.message, orchestratorApi.orchestrator); | |
} | |
} | |
module.exports = AWSHandlers; |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment