mirror of
https://github.com/kikobar/mastodon.git
synced 2024-12-25 23:34:14 +00:00
Fix crash streaming process when receive invalid json (#14859)
This commit is contained in:
parent
b09810a505
commit
e39d97f700
|
@ -81,6 +81,19 @@ const redisUrlToClient = (defaultConfig, redisUrl) => {
|
||||||
|
|
||||||
const numWorkers = +process.env.STREAMING_CLUSTER_NUM || (env === 'development' ? 1 : Math.max(os.cpus().length - 1, 1));
|
const numWorkers = +process.env.STREAMING_CLUSTER_NUM || (env === 'development' ? 1 : Math.max(os.cpus().length - 1, 1));
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @param {string} json
|
||||||
|
* @return {Object.<string, any>|null}
|
||||||
|
*/
|
||||||
|
const parseJSON = (json) => {
|
||||||
|
try {
|
||||||
|
return JSON.parse(json);
|
||||||
|
} catch (err) {
|
||||||
|
log.error(err);
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
const startMaster = () => {
|
const startMaster = () => {
|
||||||
if (!process.env.SOCKET && process.env.PORT && isNaN(+process.env.PORT)) {
|
if (!process.env.SOCKET && process.env.PORT && isNaN(+process.env.PORT)) {
|
||||||
log.warn('UNIX domain socket is now supported by using SOCKET. Please migrate from PORT hack.');
|
log.warn('UNIX domain socket is now supported by using SOCKET. Please migrate from PORT hack.');
|
||||||
|
@ -522,7 +535,9 @@ const startWorker = (workerId) => {
|
||||||
log.verbose(req.requestId, `Starting stream from ${ids.join(', ')} for ${accountId}${streamType}`);
|
log.verbose(req.requestId, `Starting stream from ${ids.join(', ')} for ${accountId}${streamType}`);
|
||||||
|
|
||||||
const listener = message => {
|
const listener = message => {
|
||||||
const { event, payload, queued_at } = JSON.parse(message);
|
const json = parseJSON(message);
|
||||||
|
if (!json) return;
|
||||||
|
const { event, payload, queued_at } = json;
|
||||||
|
|
||||||
const transmit = () => {
|
const transmit = () => {
|
||||||
const now = new Date().getTime();
|
const now = new Date().getTime();
|
||||||
|
@ -932,7 +947,9 @@ const startWorker = (workerId) => {
|
||||||
ws.on('error', onEnd);
|
ws.on('error', onEnd);
|
||||||
|
|
||||||
ws.on('message', data => {
|
ws.on('message', data => {
|
||||||
const { type, stream, ...params } = JSON.parse(data);
|
const json = parseJSON(data);
|
||||||
|
if (!json) return;
|
||||||
|
const { type, stream, ...params } = json;
|
||||||
|
|
||||||
if (type === 'subscribe') {
|
if (type === 'subscribe') {
|
||||||
subscribeWebsocketToChannel(session, firstParam(stream), params);
|
subscribeWebsocketToChannel(session, firstParam(stream), params);
|
||||||
|
|
Loading…
Reference in a new issue