-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
ft: multiple buckets and mediator registration
- Loading branch information
1 parent
9ea8276
commit bec35fc
Showing
5 changed files
with
83 additions
and
80 deletions.
There are no files selected for viewing
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
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
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,91 +1,23 @@ | ||
| import express from 'express'; | ||
| import * as Minio from 'minio'; | ||
| import path from 'path'; | ||
| import { getConfig } from './config/config'; | ||
| import logger from './logger'; | ||
| import routes from './routes/index'; | ||
| import { setupMediator } from './openhim/openhim'; | ||
| import { validateJsonFile, getCsvHeaders } from './utils/file-validators'; | ||
| import { readFile, rm } from 'fs/promises'; | ||
| import { createTable, flattenJson } from './utils/clickhouse'; | ||
| import { setupMinio } from './utils/minio'; | ||
|
|
||
| const app = express(); | ||
|
|
||
| app.use('/', routes); | ||
| const prefix = getConfig().runningMode === 'testing' ? '/' : '/climate'; | ||
|
|
||
| if (getConfig().runningMode !== 'testing') { | ||
| app.listen(getConfig().port, () => { | ||
| logger.info(`Server is running on port - ${getConfig().port}`); | ||
| app.use(prefix, routes); | ||
|
|
||
| if (getConfig().registerMediator) { | ||
| setupMediator(path.resolve(__dirname, './openhim/mediatorConfig.json')); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| async function setupMinio() { | ||
| const { bucket, endPoint, port, useSSL, accessKey, secretKey, prefix, suffix } = | ||
| getConfig().minio; | ||
| setupMinio(); | ||
|
|
||
| const minioClient = new Minio.Client({ | ||
| endPoint, | ||
| port, | ||
| useSSL, | ||
| accessKey, | ||
| secretKey, | ||
| }); | ||
| app.listen(getConfig().port, () => { | ||
| logger.info(`Server is running on port - ${getConfig().port}`); | ||
|
|
||
| try { | ||
| // Test connection by attempting to list buckets | ||
| await minioClient.listBuckets(); | ||
| logger.info(`Successfully connected to Minio at ${endPoint}:${port}/${bucket}`); | ||
| } catch (error) { | ||
| logger.error(`Failed to connect to Minio: ${error}`); | ||
| throw error; | ||
| if (getConfig().runningMode !== 'testing' && getConfig().registerMediator) { | ||
| setupMediator(path.resolve(__dirname, './openhim/mediatorConfig.json')); | ||
| } | ||
| const listener = minioClient.listenBucketNotification(bucket, prefix, suffix, [ | ||
| 's3:ObjectCreated:*', | ||
| ]); | ||
|
|
||
| listener.on('notification', async (notification) => { | ||
| //@ts-ignore | ||
| const file = notification.s3.object.key; | ||
|
|
||
| //@ts-ignore | ||
| const tableName = notification.s3.bucket.name; | ||
|
|
||
| //@ts-ignore | ||
| minioClient.fGetObject(bucket, file, `tmp/${file}`, async (err) => { | ||
| if (err) { | ||
| logger.error(err); | ||
| } else { | ||
| const fileBuffer = await readFile(`tmp/${file}`); | ||
|
|
||
| //get the file extension | ||
| const extension = file.split('.').pop(); | ||
| logger.info(`File Downloaded - Type: ${extension}`); | ||
|
|
||
| if (extension === 'json' && validateJsonFile(fileBuffer)) { | ||
| // flatten the json and pass it to clickhouse | ||
| //const fields = flattenJson(JSON.parse(fileBuffer.toString())); | ||
| //await createTable(fields, tableName); | ||
| logger.warn(`File type not currently supported- ${extension}`); | ||
| } else if (extension === 'csv' && getCsvHeaders(fileBuffer)) { | ||
| //get the first line of the csv file | ||
| const fields = (await readFile(`tmp/${file}`, 'utf8')).split('\n')[0].split(','); | ||
|
|
||
| await createTable(fields, tableName); | ||
| } else { | ||
| logger.warn(`Unknown file type - ${extension}`); | ||
| } | ||
| await rm(`tmp/${file}`); | ||
| } | ||
| }); | ||
| }); | ||
| } | ||
|
|
||
| setupMinio(); | ||
|
|
||
| app.listen(3000, () => { | ||
| logger.info(`Server is running on port - ${3000}`); | ||
| }); |
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
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| import * as Minio from 'minio'; | ||
| import { getConfig } from '../config/config'; | ||
| import logger from '../logger'; | ||
| import { readFile, rm } from 'fs/promises'; | ||
| import { createTable } from './clickhouse'; | ||
| import { validateJsonFile, getCsvHeaders } from './file-validators'; | ||
|
|
||
| export async function setupMinio() { | ||
| const { buckets, endPoint, port, useSSL, accessKey, secretKey, prefix, suffix } = | ||
| getConfig().minio; | ||
|
|
||
| const minioClient = new Minio.Client({ | ||
| endPoint, | ||
| port, | ||
| useSSL, | ||
| accessKey, | ||
| secretKey, | ||
| }); | ||
|
|
||
| try { | ||
| // Test connection by attempting to list buckets | ||
| await minioClient.listBuckets(); | ||
| logger.info(`Successfully connected to Minio at ${endPoint}:${port}`); | ||
| } catch (error) { | ||
| logger.error(`Failed to connect to Minio: ${error}`); | ||
| throw error; | ||
| } | ||
|
|
||
| const listOfBuckets = buckets.split(','); | ||
|
|
||
| for (const bucket of listOfBuckets) { | ||
| const listener = minioClient.listenBucketNotification(bucket, prefix, suffix, [ | ||
| 's3:ObjectCreated:*', | ||
| ]); | ||
|
|
||
| listener.on('notification', async (notification) => { | ||
| //@ts-ignore | ||
| const file = notification.s3.object.key; | ||
|
|
||
| //@ts-ignore | ||
| const tableName = notification.s3.bucket.name; | ||
|
|
||
| //@ts-ignore | ||
| minioClient.fGetObject(bucket, file, `tmp/${file}`, async (err) => { | ||
| if (err) { | ||
| logger.error(err); | ||
| } else { | ||
| const fileBuffer = await readFile(`tmp/${file}`); | ||
|
|
||
| //get the file extension | ||
| const extension = file.split('.').pop(); | ||
| logger.info(`File Downloaded - Type: ${extension}`); | ||
|
|
||
| if (extension === 'json' && validateJsonFile(fileBuffer)) { | ||
| // flatten the json and pass it to clickhouse | ||
| //const fields = flattenJson(JSON.parse(fileBuffer.toString())); | ||
| //await createTable(fields, tableName); | ||
| logger.warn(`File type not currently supported- ${extension}`); | ||
| } else if (extension === 'csv' && getCsvHeaders(fileBuffer)) { | ||
| //get the first line of the csv file | ||
| const fields = (await readFile(`tmp/${file}`, 'utf8')).split('\n')[0].split(','); | ||
|
|
||
| await createTable(fields, tableName); | ||
| } else { | ||
| logger.warn(`Unknown file type - ${extension}`); | ||
| } | ||
| await rm(`tmp/${file}`); | ||
| } | ||
| }); | ||
| }); | ||
| } | ||
| } |