oxmc/myDrive
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

started to fix video streaming

Browse Source
This commit is contained in:
kyle hoell
2020-11-27 12:16:55 -05:00
parent 1a3930b8da
commit 1764a8c01c
6 changed files with 251 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/services/ChunkService/FileSystemService.ts
+67 -12
View File
@@ -27,6 +27,7 @@ import fixStartChunkLength from "./utils/fixStartChunkLength";
import Folder, { FolderInterface } from "../../models/folder";
import addToStoageSize from "./utils/addToStorageSize";
import subtractFromStorageSize from "./utils/subtractFromStorageSize";
const StreamSkip = require("stream-skip");
const dbUtilsFile = new DbUtilFile();
@@ -288,14 +289,25 @@ class FileSystemService implements ChunkInterface {
let currentIV = IV;
let fixedStart = start % 16 === 0 ? start : fixStartChunkLength(start);
//let fixedStart = start % 16 === 0 ? start : fixStartChunkLength(start);
let fixedStart = 0;
let fixedEnd = currentFile.length;
if (start === 0 && end === 1) {
console.log("safari request");
fixedStart = 0;
fixedEnd = 15;
} else {
fixedStart = start % 16 === 0 ? start : fixStartChunkLength(start);
}
if (+start === 0) {
fixedStart = 0;
}
const fixedEnd = currentFile.length;
const differenceStart = start - fixedStart;
@@ -303,12 +315,45 @@ class FileSystemService implements ChunkInterface {
currentIV = await getPrevIVFS(fixedStart - 16, currentFile.metadata.filePath!) as Buffer;
}
const readStream = fs.createReadStream(currentFile.metadata.filePath!, {
start: fixedStart,
end: fixedEnd,
highWaterMark: 128 * 1024
});
console.log("start", start, fixedStart, end, fixedEnd, differenceStart);
let readStream;
if (start === 0 && end === 1) {
readStream = fs.createReadStream(currentFile.metadata.filePath!, {
start: fixedStart,
end: fixedEnd,
});
} else {
readStream = fs.createReadStream(currentFile.metadata.filePath!, {
start: fixedStart,
end: fixedEnd,
// highWaterMark: 1024
})
}
// } else if (fixedStart !== 0 && start !== 0) {
// readStream = fs.createReadStream(currentFile.metadata.filePath!, {
// start: fixedStart + differenceStart,
// end: fixedEnd,
// });
// } else {
// readStream = fs.createReadStream(currentFile.metadata.filePath!, {
// start: fixedStart,
// end: fixedEnd,
// });
// }
// const readStream = fs.createReadStream(currentFile.metadata.filePath!, {
// start: fixedStart,
// end: fixedEnd,
// });
const CIPHER_KEY = crypto.createHash('sha256').update(password).digest()
@@ -322,9 +367,9 @@ class FileSystemService implements ChunkInterface {
readStream.pipe(decipher);
readStream.on("close", () => {
console.log("read stream closed")
})
// readStream.on("close", () => {
// console.log("read stream closed")
// })
// req.on("close", () => {
// // console.log("req closed");
@@ -335,7 +380,17 @@ class FileSystemService implements ChunkInterface {
//console.log("temp uuid", tempUUID);
await awaitStreamVideo(start, end, differenceStart, decipher, res, req, tempUUID, allStreamsToErrorCatch);
// return;
// if (start === 0 && end === 1) {
// await awaitStreamVideo(start, end, differenceStart, decipher, res, req, tempUUID, allStreamsToErrorCatch);
// } else {
// await awaitStreamVideo(start, end, differenceStart, decipher, res, req, tempUUID, allStreamsToErrorCatch);
// }
await awaitStreamVideo(start, end, differenceStart, decipher, res, req, tempUUID, allStreamsToErrorCatch, readStream);
console.log("await stream finished")
readStream.destroy();
}
backend/services/ChunkService/MongoService.ts
+1 -1
View File
@@ -349,7 +349,7 @@ class MongoService implements ChunkInterface {
const tempUUID = req.params.uuid;
await awaitStreamVideo(start, end, differenceStart, decipher, res, req, tempUUID, allStreamsToErrorCatch);
//await awaitStreamVideo(start, end, differenceStart, decipher, res, req, tempUUID, allStreamsToErrorCatch);
readStream.destroy();
}
backend/services/ChunkService/S3Service.ts
+1 -1
View File
@@ -307,7 +307,7 @@ class S3Service implements ChunkInterface {
console.log("temp uuid", tempUUID);
await awaitStreamVideo(start, end, differenceStart, decipher, res, req, tempUUID, allStreamsToErrorCatch);
//await awaitStreamVideo(start, end, differenceStart, decipher, res, req, tempUUID, allStreamsToErrorCatch);
s3ReadStream.destroy();
}
backend/services/ChunkService/utils/awaitStreamVideo.ts
+120 -84
View File
@@ -1,9 +1,10 @@
import {Response, Request} from "express"
import tempStorage from "../../../tempStorage/tempStorage";
import uuid from "uuid";
const StreamSkip = require("stream-skip");
const awaitStreamVideo = (start: number, end:number, differenceStart: number,
decipher: any, res: Response, req: Request, tempUUID: string, streamsToErrorCatch: any[]) => {
decipher: any, res: Response, req: Request, tempUUID: string, streamsToErrorCatch: any[], readStream: any) => {
// const currentUUID = uuid.v4();
// tempStorage[tempUUID] = currentUUID;
@@ -15,9 +16,11 @@ const awaitStreamVideo = (start: number, end:number, differenceStart: number,
req.on("close", () => {
console.log("close resolved");
streamsToErrorCatch.forEach((stream) => {
stream.destroy();
})
// streamsToErrorCatch.forEach((stream) => {
// stream.destroy();
// })
readStream.close();
decipher.destroy();
resolve();
})
@@ -29,97 +32,130 @@ const awaitStreamVideo = (start: number, end:number, differenceStart: number,
resolve();
})
decipher.on("data", (data: Buffer | string) => {
readStream.on("close", () => {
console.log("read stream closed")
decipher.destroy();
})
if (+start === 0 && +end === 1) {
console.log("safari first byte request");
//console.log("data", uuid.v4());
decipher.on("data", (data: Buffer | string) => {
//console.log("original data size", data.length);
// if (tempStorage[tempUUID] !== currentUUID) {
// streamsToErrorCatch.forEach((stream) => {
// stream.destroy();
// })
// resolve();
// }
//console.log("data", currentUUID);
//console.log("stream passed", currentUUID);
// if (tempStorage[tempUUID] !== undefined && tempStorage[tempUUID] !== currentUUID) {
// console.log("New Stream Requested, Desroying old stream");
// streamsToErrorCatch[0].destroy();
// console.log("Old Stream Desroyed");
// delete tempStorage[tempUUID];
// } else {
// tempStorage[tempUUID] = currentUUID;
// }
console.log("data size", data.length)
if (+start === 0 && +end === 1) {
const dataCoverted = data.toString("hex");
let neededData = dataCoverted.substring(0, 4);
let neededData = dataCoverted.substring(0, 8);
const dataBack = Buffer.from(neededData, "hex");
res.write(dataBack);
//res.flush();
return;
}
if (!firstBytesRemoved) {
const dataCoverted = data.toString("hex");
let neededData = dataCoverted.substring(differenceStart * 2);
const dataBack = Buffer.from(neededData, "hex");
firstBytesRemoved = true;
//sizeCounter += dataBack.length;
//console.log("safari bytes size", dataBack.length);
res.write(dataBack);
//res.flush();
return;
}
res.write(data);
//res.flush();
//sizeCounter += data.length;
})
decipher.on("close", () => {
console.log("decipher closed");
})
decipher.on("end", () => {
console.log("decipher resolved");
res.end();
//resolve();
})
streamsToErrorCatch.forEach((currentStream) => {
currentStream.on("error", (e: Error) => {
reject({
message: "Await Video Stream Input Error",
code: 500,
error: e
})
})
})
} else {
const streamSkip = new StreamSkip({skip: differenceStart});
decipher.pipe(streamSkip).pipe(res);
}
//decipher.pipe(res);
// decipher.on("data", (data: Buffer | string) => {
// //console.log("data", uuid.v4());
// // if (tempStorage[tempUUID] !== currentUUID) {
// // streamsToErrorCatch.forEach((stream) => {
// // stream.destroy();
// // })
// // resolve();
// // }
// //console.log("data", currentUUID);
// //console.log("stream passed", currentUUID);
// // if (tempStorage[tempUUID] !== undefined && tempStorage[tempUUID] !== currentUUID) {
// // console.log("New Stream Requested, Desroying old stream");
// // streamsToErrorCatch[0].destroy();
// // console.log("Old Stream Desroyed");
// // delete tempStorage[tempUUID];
// // } else {
// // tempStorage[tempUUID] = currentUUID;
// // }
// //console.log("data size", data.length)
// if (+start === 0 && +end === 1) {
// const dataCoverted = data.toString("hex");
// let neededData = dataCoverted.substring(0, 4);
// const dataBack = Buffer.from(neededData, "hex");
// res.write(dataBack);
// //res.flush();
// // return;
// } else if (!firstBytesRemoved) {
// const dataCoverted = data.toString("hex");
// let neededData = dataCoverted.substring(differenceStart * 2);
// const dataBack = Buffer.from(neededData, "hex");
// firstBytesRemoved = true;
// //sizeCounter += dataBack.length;
// res.write(dataBack);
// //res.flush();
// // return;
// } else {
// res.write(data);
// }
// //res.flush();
// //sizeCounter += data.length;
// })
// decipher.on("close", () => {
// console.log("decipher closed");
// })
// decipher.on("end", () => {
// console.log("decipher resolved");
// //res.end();
// //resolve();
// })
// streamsToErrorCatch.forEach((currentStream) => {
// currentStream.on("error", (e: Error) => {
// reject({
// message: "Await Video Stream Input Error",
// code: 500,
// error: e
// })
// })
// })
})
}
package.json
+2
View File
@@ -17,6 +17,7 @@
"dev-server": "webpack-dev-server",
"test": "NODE_ENV=test env-cmd -f ./config/test.env jest --config=jest.config.json --watchAll --runInBand",
"start:dev": "NODE_ENV=development node dist/server/serverStart.js",
"start:debug": "NODE_ENV=development node --inspect dist/server/serverStart.js",
"start": "node dist/server/serverStart.js --env production",
"create-indexes-database": "node serverUtils/createIndexes.js",
"increase-node-memory": "export NODE_OPTIONS=--max_old_space_size=8000",
@@ -101,6 +102,7 @@
"request": "^2.88.2",
"request-ip": "^2.1.3",
"sharp": "^0.23.4",
"stream-skip": "^1.0.3",
"sweetalert2": "^9.7.2",
"temp": "^0.9.1",
"typescript": "^3.8.3",
src/axiosInterceptor/index.js
+60 -145
View File
@@ -1,109 +1,58 @@
import axios from 'axios';
import createAuthRefreshInterceptor from "axios-auth-refresh";
// const refreshAuthLogic = failedRequest => axios.post("/user-service/get-token").then(tokenRefreshResponse => {
// return Promise.resolve();
// }).catch(() => {
// return Promise.reject(failedRequest);
// });
// // Instantiate the interceptor (you can chain it as it returns the axios instance)
// createAuthRefreshInterceptor(axios, refreshAuthLogic);
const sleep = () => {
return new Promise((resolve, reject) => {
setTimeout(() => {
resolve();
}, 150);
})
}
// const client = axios.create();
// axiosRetry(client, {
// retries: 0,
// })
//axiosRetry(axios, { retries: 3 });
// const getNewToken = () => {
// return new Promise((resolve, reject) => {
// axios.post("/user-service/get-token").then((response) => {
// console.log("get token response");
// resolve();
// }).catch((e) => {
// resolve();
// })
// })
// }
// axios.interceptors.response.use(null, (error) => {
// if (error.config && error.response && error.response.status === 401 && !error._tried) {
// return getNewToken().then(() => {
// console.log("got new token");
// error._tried = true;
// return axios.request(error.config);
// })
// }
// return Promise.reject(error);
// });
const axiosNoRetry = axios.create();
const axiosRetry = axios.create();
const axiosNoRetry = axios.create();
const axios3 = axios.create();
const sleep = (sleepybio) => {
return new Promise((resolve, reject) => {
setTimeout(() => {
resolve(sleepybio);
}, 100);
})
}
const sleep2 = (sleepybio) => {
return new Promise((resolve, reject) => {
setTimeout(() => {
resolve(sleepybio);
}, 1500);
})
}
axiosRetry.interceptors.response.use((response) => {
console.log("axios interceptor successful")
//console.log("axios interceptor successful")
return response;
}, (error) => {
return new Promise((resolve, reject) => {
console.log("request interceptor failed", error.config.url);
//console.log("request interceptor failed", error.config.url);
let originalRequest = error.config;
if (error.response.status !== 401) {
//console.log("error does not equal 401");
return reject(error);
}
if (originalRequest.ran === true) {
console.log("original request ran", error.config.url);
//console.log("original request ran", error.config.url);
return reject(error);
}
if (error.config.url === "/user-service/get-token") {
console.log("error url equal to refresh token route")
//console.log("error url equal to refresh token route")
return reject();
}
}
axiosNoRetry.post("/user-service/get-token").then((cookieResponse) => {
console.log("sleeping")
return sleep("sleepy boi");
}).then((sleepres) => {
console.log(sleepres);
console.time("test");
return axios3(originalRequest);
}).then((response) => {
console.timeEnd("test");
console.log("axios 3");
resolve(response)
}).catch((e) => {
console.log("error");
//console.log("error");
return reject(error);
})
@@ -114,95 +63,61 @@ axiosRetry.interceptors.response.use((response) => {
// axios.interceptors.response.use( (response) => {
// console.log("axios interceptor")
// // Return a successful response back to the calling service
// return response;
// }, (error) => {
// const originalRequest = error.config;
// if (originalRequest.ran) {
// console.log("Request already ran");
// return Promise.reject(error);
// }
// console.log("axios interceptor failed first request")
// // Return any error which is not due to authentication back to the calling service
// if (error.response.status !== 401) {
// console.log("error does not equal 401");
// return new Promise((resolve, reject) => {
// reject(error);
// });
// }
// //axios.interceptors.response.eject(this);
// // Logout user if token refresh didn't work or user is disabled
// const originalRequest = error.config;
// console.log("error url", error.config.url);
// if (originalRequest.ran === 3) {
// console.log("Max retry count", error.config.url);
// return reject(error);
// }
// if (error.config.url === "/user-service/get-token") {
// console.log("error url equal to refresh token route")
// return new Promise((resolve, reject) => {
// reject(error);
// });
// }
// console.log("axios retry count", originalRequest.ran, originalRequest.ran === 3);
// // Try request again with new token
// if (error.response.status !== 401) {
// console.log("error does not equal 401");
// return reject(error);
// }
// // return new Promise((resolve, reject) => {
// // })
// axiosNoRetry.post("/user-service/get-token").then((cookieResponse) => {
// return axios.post("/user-service/get-token").then((cookieResponse) => {
// console.log("cookie status", cookieResponse.status);
// console.log("cookie status", cookieResponse.status);
// if (cookieResponse.status === 201) {
// originalRequest.ran = originalRequest.ran ? originalRequest.ran + 1 : 1;
// //return axios(originalRequest);
// //return resolve(axios(originalRequest));
// if (cookieResponse.status === 201) {
// console.log("cookie response", cookieResponse.data);
// return "value"
// originalRequest.ran = true;
// return axios(originalRequest);
// } else {
// return reject(error);
// }
// }).then(() => {
// axiosNoRetry(originalRequest).then((newRequest) => {
// console.log("new request", newRequest.data);
// resolve(newRequest)
// }).catch((newError) => {
// console.log("new request error", newError);
// return reject(newError)
// })
// })
// })
// // const originalRequest = error.config;
// // if (originalRequest.ran === 3) {
// // console.log("Max retry count", error.config.url);
// // return Promise.reject(error);
// // }
// // console.log("axios retry count", originalRequest.ran);
// // if (error.response.status !== 401) {
// // console.log("error does not equal 401");
// // return new Promise((resolve, reject) => {
// // reject(error);
// // });
// // }
// // if (error.config.url === "/user-service/get-token") {
// // console.log("error url equal to refresh token route")
// // return new Promise((resolve, reject) => {
// // reject(error);
// // });
// // }
// // return axiosNoRetry.post("/user-service/get-token").then((cookieResponse) => {
// // console.log("cookie status", cookieResponse.status);
// // if (cookieResponse.status === 201) {
// // originalRequest.ran = originalRequest.ran ? originalRequest.ran + 1 : 1;
// // return axios(originalRequest);
// // } else {
// // return Promise.reject(error);
// // }
// // })
// } else {
// return Promise.reject(error);
// }
// })
// // .catch((cookieError) => {
// // console.log("cookieError", cookieError);
// // return Promise.reject(error);
// // })
// });
export default axiosRetry;