forked from Sunbird-Obsrv/obsrv-api-service
-
Notifications
You must be signed in to change notification settings - Fork 9
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #196 from Sanketika-Obsrv/connector-registry-stream
Connector registry stream upload
- Loading branch information
Showing
8 changed files
with
207 additions
and
58 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
121 changes: 121 additions & 0 deletions
121
api-service/src/v2/controllers/ConnectorRegister/ConnectorRegisterController.ts
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,121 @@ | ||
| import { Request, Response } from "express"; | ||
| import { ResponseHandler } from "../../helpers/ResponseHandler"; | ||
| import _ from "lodash"; | ||
| import logger from "../../logger"; | ||
| import axios from "axios"; | ||
| import httpStatus from "http-status"; | ||
| import busboy from "busboy"; | ||
| import { PassThrough } from "stream"; | ||
| import { generatePreSignedUrl } from "../GenerateSignedURL/helper"; | ||
| import { registerConnector } from "../../connections/commandServiceConnection"; | ||
|
|
||
| export const apiId = "api.connector.register"; | ||
| export const code = "FAILED_TO_REGISTER_CONNECTOR"; | ||
|
|
||
| let resmsgid: string | any; | ||
|
|
||
| const connectorRegisterController = async (req: Request, res: Response) => { | ||
| resmsgid = _.get(res, "resmsgid"); | ||
| try { | ||
| const uploadStreamResponse: any = await uploadStream(req); | ||
| const payload = { | ||
| relative_path: uploadStreamResponse[0] | ||
| } | ||
| logger.info({ apiId, resmsgid, message: `File uploaded to cloud provider successfully` }) | ||
| const registryResponse = await registerConnector(payload); | ||
| logger.info({ apiId, resmsgid, message: `Connector registered successfully` }) | ||
| ResponseHandler.successResponse(req, res, { status: httpStatus.OK, data: { message: registryResponse?.data?.message } }) | ||
| } catch (error: any) { | ||
| const errMessage = _.get(error, "response.data.error.message") | ||
| logger.error(error, apiId, resmsgid, code); | ||
| let errorMessage = error; | ||
| const statusCode = _.get(error, "statusCode") | ||
| if (!statusCode || statusCode == 500) { | ||
| errorMessage = { code, message: errMessage || "Failed to register connector" } | ||
| } | ||
| ResponseHandler.errorResponse(errorMessage, req, res); | ||
| } | ||
| }; | ||
|
|
||
| const uploadStream = async (req: Request) => { | ||
| return new Promise((resolve, reject) => { | ||
| const filePromises: Promise<void>[] = []; | ||
| const busboyClient = busboy({ headers: req.headers }); | ||
| const relative_path: any[] = []; | ||
| let fileCount = 0; | ||
|
|
||
| busboyClient.on("file", async (name: any, file: any, info: any) => { | ||
| if (fileCount > 0) { | ||
| // If more than one file is detected, reject the request | ||
| busboyClient.emit("error", reject({ | ||
| code: "FAILED_TO_UPLOAD", | ||
| message: "Uploading multiple files are not allowed", | ||
| statusCode: 400, | ||
| errCode: "BAD_REQUEST" | ||
| })); | ||
| return | ||
| } | ||
| fileCount++; | ||
| const processFile = async () => { | ||
| const fileName = info?.filename; | ||
| try { | ||
| const preSignedUrl: any = await generatePreSignedUrl("write", [fileName], "connector") | ||
| const filePath = preSignedUrl[0]?.filePath | ||
| const fileNameExtracted = extractFileNameFromPath(filePath); | ||
| relative_path.push(...fileNameExtracted); | ||
| const pass = new PassThrough(); | ||
| file.pipe(pass); | ||
| const fileBuffer = await streamToBuffer(pass); | ||
| await axios.put(preSignedUrl[0]?.preSignedUrl, fileBuffer, { | ||
| headers: { | ||
| "Content-Type": info.mimeType, | ||
| "Content-Length": fileBuffer.length, | ||
| } | ||
| }); | ||
| } | ||
| catch (err) { | ||
| logger.error({ apiId, err, resmsgid, message: "Failed to generate sample urls", code: "FILES_GENERATE_URL_FAILURE" }) | ||
| reject({ | ||
| code: "FILES_GENERATE_URL_FAILURE", | ||
| message: "Failed to generate sample urls", | ||
| statusCode: 500, | ||
| errCode: "INTERNAL_SERVER_ERROR" | ||
| }) | ||
| } | ||
| }; | ||
| filePromises.push(processFile()); | ||
| }); | ||
| busboyClient.on("close", async () => { | ||
| try { | ||
| await Promise.all(filePromises); | ||
| resolve(relative_path); | ||
| } catch (error) { | ||
| logger.error({ apiId, error, resmsgid, message: "Fail to upload a file", code: "FAILED_TO_UPLOAD" }) | ||
| reject({ | ||
| code: "FAILED_TO_UPLOAD", | ||
| message: "Fail to upload a file", | ||
| statusCode: 400, | ||
| errCode: "BAD_REQUEST" | ||
| }); | ||
| } | ||
| }); | ||
| busboyClient.on("error", reject); | ||
| req.pipe(busboyClient); | ||
| }) | ||
| } | ||
|
|
||
| const streamToBuffer = (stream: PassThrough): Promise<Buffer> => { | ||
| return new Promise<Buffer>((resolve, reject) => { | ||
| const chunks: Buffer[] = []; | ||
| stream.on("data", (chunk) => chunks.push(chunk)); | ||
| stream.on("end", () => resolve(Buffer.concat(chunks))); | ||
| stream.on("error", reject); | ||
| }); | ||
| }; | ||
|
|
||
| const extractFileNameFromPath = (filePath: string): string[] => { | ||
| const regex = /(?<=\/)[^/]+\.[^/]+(?=\/|$)/g; | ||
| return filePath.match(regex) || []; | ||
| }; | ||
|
|
||
| export default connectorRegisterController; |
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
64 changes: 64 additions & 0 deletions
64
api-service/src/v2/controllers/GenerateSignedURL/helper.ts
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,64 @@ | ||
| import { config } from "../../configs/Config"; | ||
| import * as _ from "lodash"; | ||
| import { cloudProvider } from "../../services/CloudServices"; | ||
| import { URLAccess } from "../../types/SampleURLModel"; | ||
| import path from "path"; | ||
| import { v4 as uuidv4 } from "uuid"; | ||
|
|
||
| export const generatePreSignedUrl = async (access: string, files: any, containerType: string) => { | ||
| const { filesList, updatedFileNames } = transformFileNames(files, access, containerType); | ||
| const urlExpiry: number = getURLExpiry(access); | ||
| const preSignedUrls = await Promise.all(cloudProvider.generateSignedURLs(config.cloud_config.container, updatedFileNames, access, urlExpiry)); | ||
| const signedUrlList = _.map(preSignedUrls, list => { | ||
| const fileNameWithUid = _.keys(list)[0]; | ||
| return { | ||
| filePath: getFilePath(fileNameWithUid, containerType), | ||
| fileName: filesList.get(fileNameWithUid), | ||
| preSignedUrl: _.values(list)[0] | ||
| }; | ||
| }); | ||
| return signedUrlList; | ||
| } | ||
|
|
||
| const getFilePath = (file: string, containerType: string) => { | ||
| const datasetUploadPath = `${config.presigned_url_configs.service}/user_uploads/${file}`; | ||
| const connectorUploadPath = `${config.cloud_config.container_prefix}/${file}`; | ||
| const paths: Record<string, string> = { | ||
| "dataset": datasetUploadPath, | ||
| "connector": connectorUploadPath | ||
| }; | ||
| return paths[containerType] || datasetUploadPath; | ||
| } | ||
|
|
||
| const transformFileNames = (fileList: Array<string | any>, access: string, containerType: string): Record<string, any> => { | ||
| if (access === URLAccess.Read) { | ||
| return transformReadFiles(fileList, containerType); | ||
| } | ||
| return transformWriteFiles(fileList, containerType); | ||
| }; | ||
|
|
||
| const transformReadFiles = (fileNames: Array<string | any>, containerType: string) => { | ||
| const fileMap = new Map(); | ||
| const updatedFileNames = fileNames.map(file => { | ||
| fileMap.set(file, file); | ||
| return getFilePath(file, containerType); | ||
| }); | ||
| return { filesList: fileMap, updatedFileNames }; | ||
| }; | ||
|
|
||
| const transformWriteFiles = (fileNames: Array<string | any>, containerType: string) => { | ||
| const fileMap = new Map(); | ||
| const updatedFileNames = fileNames.map(file => { | ||
| const uuid = uuidv4().replace(/-/g, "").slice(0, 6); | ||
| const ext = path.extname(file); | ||
| const baseName = path.basename(file, ext); | ||
| const updatedFileName = `${baseName}_${uuid}${ext}`; | ||
| fileMap.set(updatedFileName, file); | ||
| return getFilePath(updatedFileName, containerType); | ||
| }); | ||
| return { filesList: fileMap, updatedFileNames }; | ||
| }; | ||
|
|
||
| const getURLExpiry = (access: string) => { | ||
| return access === URLAccess.Read ? config.presigned_url_configs.read_storage_url_expiry : config.presigned_url_configs.write_storage_url_expiry; | ||
| } |
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