Add no-misused/floating-promises lint rule
This commit is contained in:
Fedor Indutny
GitHub
parent
1a68c3db62
commit
ed271d92ea
150 changed files with 1296 additions and 991 deletions
|
|
@ -40,6 +40,7 @@ import { verifySignature } from '../Curve';
|
|||
import { strictAssert } from '../util/assert';
|
||||
import type { BatcherType } from '../util/batcher';
|
||||
import { createBatcher } from '../util/batcher';
|
||||
import { drop } from '../util/drop';
|
||||
import { dropNull } from '../util/dropNull';
|
||||
import { normalizeUuid } from '../util/normalizeUuid';
|
||||
import { parseIntOrThrow } from '../util/parseIntOrThrow';
|
||||
|
|
@ -312,7 +313,7 @@ export default class MessageReceiver
|
|||
processBatch: (items: Array<CacheAddItemType>) => {
|
||||
// Not returning the promise here because we don't want to stall
|
||||
// the batch.
|
||||
this.decryptAndCacheBatch(items);
|
||||
void this.decryptAndCacheBatch(items);
|
||||
},
|
||||
});
|
||||
this.cacheRemoveBatcher = createBatcher<string>({
|
||||
|
|
@ -337,13 +338,15 @@ export default class MessageReceiver
|
|||
request.respond(200, 'OK');
|
||||
|
||||
if (request.verb === 'PUT' && request.path === '/api/v1/queue/empty') {
|
||||
this.incomingQueue.add(
|
||||
createTaskWithTimeout(
|
||||
async () => {
|
||||
this.onEmpty();
|
||||
},
|
||||
'incomingQueue/onEmpty',
|
||||
TASK_WITH_TIMEOUT_OPTIONS
|
||||
drop(
|
||||
this.incomingQueue.add(
|
||||
createTaskWithTimeout(
|
||||
async () => {
|
||||
this.onEmpty();
|
||||
},
|
||||
'incomingQueue/onEmpty',
|
||||
TASK_WITH_TIMEOUT_OPTIONS
|
||||
)
|
||||
)
|
||||
);
|
||||
}
|
||||
|
|
@ -422,22 +425,26 @@ export default class MessageReceiver
|
|||
}
|
||||
};
|
||||
|
||||
this.incomingQueue.add(
|
||||
createTaskWithTimeout(
|
||||
job,
|
||||
'incomingQueue/websocket',
|
||||
TASK_WITH_TIMEOUT_OPTIONS
|
||||
drop(
|
||||
this.incomingQueue.add(
|
||||
createTaskWithTimeout(
|
||||
job,
|
||||
'incomingQueue/websocket',
|
||||
TASK_WITH_TIMEOUT_OPTIONS
|
||||
)
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
public reset(): void {
|
||||
// We always process our cache before processing a new websocket message
|
||||
this.incomingQueue.add(
|
||||
createTaskWithTimeout(
|
||||
async () => this.queueAllCached(),
|
||||
'incomingQueue/queueAllCached',
|
||||
TASK_WITH_TIMEOUT_OPTIONS
|
||||
drop(
|
||||
this.incomingQueue.add(
|
||||
createTaskWithTimeout(
|
||||
async () => this.queueAllCached(),
|
||||
'incomingQueue/queueAllCached',
|
||||
TASK_WITH_TIMEOUT_OPTIONS
|
||||
)
|
||||
)
|
||||
);
|
||||
|
||||
|
|
@ -626,11 +633,13 @@ export default class MessageReceiver
|
|||
//
|
||||
|
||||
private async dispatchAndWait(id: string, event: Event): Promise<void> {
|
||||
this.appQueue.add(
|
||||
createTaskWithTimeout(
|
||||
async () => Promise.all(this.dispatchEvent(event)),
|
||||
`dispatchEvent(${event.type}, ${id})`,
|
||||
TASK_WITH_TIMEOUT_OPTIONS
|
||||
drop(
|
||||
this.appQueue.add(
|
||||
createTaskWithTimeout(
|
||||
async () => Promise.all(this.dispatchEvent(event)),
|
||||
`dispatchEvent(${event.type}, ${id})`,
|
||||
TASK_WITH_TIMEOUT_OPTIONS
|
||||
)
|
||||
)
|
||||
);
|
||||
}
|
||||
|
|
@ -707,16 +716,24 @@ export default class MessageReceiver
|
|||
);
|
||||
|
||||
// We don't await here because we don't want this to gate future message processing
|
||||
this.appQueue.add(
|
||||
createTaskWithTimeout(emitEmpty, 'emitEmpty', TASK_WITH_TIMEOUT_OPTIONS)
|
||||
drop(
|
||||
this.appQueue.add(
|
||||
createTaskWithTimeout(
|
||||
emitEmpty,
|
||||
'emitEmpty',
|
||||
TASK_WITH_TIMEOUT_OPTIONS
|
||||
)
|
||||
)
|
||||
);
|
||||
};
|
||||
|
||||
const waitForEncryptedQueue = async () => {
|
||||
this.addToQueue(
|
||||
waitForDecryptedQueue,
|
||||
'onEmpty/waitForDecrypted',
|
||||
TaskType.Decrypted
|
||||
drop(
|
||||
this.addToQueue(
|
||||
waitForDecryptedQueue,
|
||||
'onEmpty/waitForDecrypted',
|
||||
TaskType.Decrypted
|
||||
)
|
||||
);
|
||||
};
|
||||
|
||||
|
|
@ -725,25 +742,29 @@ export default class MessageReceiver
|
|||
// Resetting count so everything from the websocket after this starts at zero
|
||||
this.count = 0;
|
||||
|
||||
this.addToQueue(
|
||||
waitForEncryptedQueue,
|
||||
'onEmpty/waitForEncrypted',
|
||||
TaskType.Encrypted
|
||||
drop(
|
||||
this.addToQueue(
|
||||
waitForEncryptedQueue,
|
||||
'onEmpty/waitForEncrypted',
|
||||
TaskType.Encrypted
|
||||
)
|
||||
);
|
||||
};
|
||||
|
||||
const waitForCacheAddBatcher = async () => {
|
||||
await this.decryptAndCacheBatcher.onIdle();
|
||||
this.incomingQueue.add(
|
||||
createTaskWithTimeout(
|
||||
waitForIncomingQueue,
|
||||
'onEmpty/waitForIncoming',
|
||||
TASK_WITH_TIMEOUT_OPTIONS
|
||||
drop(
|
||||
this.incomingQueue.add(
|
||||
createTaskWithTimeout(
|
||||
waitForIncomingQueue,
|
||||
'onEmpty/waitForIncoming',
|
||||
TASK_WITH_TIMEOUT_OPTIONS
|
||||
)
|
||||
)
|
||||
);
|
||||
};
|
||||
|
||||
waitForCacheAddBatcher();
|
||||
drop(waitForCacheAddBatcher());
|
||||
}
|
||||
|
||||
private updateProgress(count: number): void {
|
||||
|
|
@ -835,15 +856,20 @@ export default class MessageReceiver
|
|||
};
|
||||
|
||||
// Maintain invariant: encrypted queue => decrypted queue
|
||||
this.addToQueue(
|
||||
async () => {
|
||||
this.queueDecryptedEnvelope(decryptedEnvelope, payloadPlaintext);
|
||||
},
|
||||
`queueDecryptedEnvelope(${getEnvelopeId(decryptedEnvelope)})`,
|
||||
TaskType.Encrypted
|
||||
drop(
|
||||
this.addToQueue(
|
||||
async () => {
|
||||
void this.queueDecryptedEnvelope(
|
||||
decryptedEnvelope,
|
||||
payloadPlaintext
|
||||
);
|
||||
},
|
||||
`queueDecryptedEnvelope(${getEnvelopeId(decryptedEnvelope)})`,
|
||||
TaskType.Encrypted
|
||||
)
|
||||
);
|
||||
} else {
|
||||
this.queueCachedEnvelope(item, envelope);
|
||||
void this.queueCachedEnvelope(item, envelope);
|
||||
}
|
||||
} catch (error) {
|
||||
log.error(
|
||||
|
|
@ -876,11 +902,13 @@ export default class MessageReceiver
|
|||
if (this.isEmptied) {
|
||||
this.clearRetryTimeout();
|
||||
this.retryCachedTimeout = setTimeout(() => {
|
||||
this.incomingQueue.add(
|
||||
createTaskWithTimeout(
|
||||
async () => this.queueAllCached(),
|
||||
'queueAllCached',
|
||||
TASK_WITH_TIMEOUT_OPTIONS
|
||||
drop(
|
||||
this.incomingQueue.add(
|
||||
createTaskWithTimeout(
|
||||
async () => this.queueAllCached(),
|
||||
'queueAllCached',
|
||||
TASK_WITH_TIMEOUT_OPTIONS
|
||||
)
|
||||
)
|
||||
);
|
||||
}, RETRY_TIMEOUT);
|
||||
|
|
@ -1155,11 +1183,13 @@ export default class MessageReceiver
|
|||
logId = getEnvelopeId(unsealedEnvelope);
|
||||
|
||||
const taskId = `dispatchEvent(EnvelopeEvent(${logId}))`;
|
||||
this.addToQueue(
|
||||
async () =>
|
||||
this.dispatchAndWait(taskId, new EnvelopeEvent(unsealedEnvelope)),
|
||||
taskId,
|
||||
TaskType.Decrypted
|
||||
drop(
|
||||
this.addToQueue(
|
||||
async () =>
|
||||
this.dispatchAndWait(taskId, new EnvelopeEvent(unsealedEnvelope)),
|
||||
taskId,
|
||||
TaskType.Decrypted
|
||||
)
|
||||
);
|
||||
|
||||
return this.decryptEnvelope(stores, unsealedEnvelope, uuidKind);
|
||||
|
|
@ -1887,10 +1917,12 @@ export default class MessageReceiver
|
|||
);
|
||||
|
||||
// Avoid deadlocks by scheduling processing on decrypted queue
|
||||
this.addToQueue(
|
||||
async () => this.dispatchEvent(event),
|
||||
`decrypted/dispatchEvent/DecryptionErrorEvent(${envelopeId})`,
|
||||
TaskType.Decrypted
|
||||
drop(
|
||||
this.addToQueue(
|
||||
async () => this.dispatchEvent(event),
|
||||
`decrypted/dispatchEvent/DecryptionErrorEvent(${envelopeId})`,
|
||||
TaskType.Decrypted
|
||||
)
|
||||
);
|
||||
} else {
|
||||
this.removeFromCache(envelope);
|
||||
|
|
@ -2099,7 +2131,7 @@ export default class MessageReceiver
|
|||
},
|
||||
this.removeFromCache.bind(this, envelope)
|
||||
);
|
||||
this.dispatchAndWait(logId, ev);
|
||||
void this.dispatchAndWait(logId, ev);
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -2158,7 +2190,7 @@ export default class MessageReceiver
|
|||
},
|
||||
this.removeFromCache.bind(this, envelope)
|
||||
);
|
||||
this.dispatchAndWait(logId, ev);
|
||||
void this.dispatchAndWait(logId, ev);
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
|
@ -2828,7 +2860,7 @@ export default class MessageReceiver
|
|||
}
|
||||
|
||||
if (sentMessage.storyMessage) {
|
||||
this.handleStoryMessage(
|
||||
void this.handleStoryMessage(
|
||||
envelope,
|
||||
sentMessage.storyMessage,
|
||||
sentMessage
|
||||
|
|
@ -2867,7 +2899,7 @@ export default class MessageReceiver
|
|||
return this.handleContacts(envelope, syncMessage.contacts);
|
||||
}
|
||||
if (syncMessage.groups) {
|
||||
this.handleGroups(envelope, syncMessage.groups);
|
||||
void this.handleGroups(envelope, syncMessage.groups);
|
||||
return;
|
||||
}
|
||||
if (syncMessage.blocked) {
|
||||
|
|
@ -3333,7 +3365,7 @@ export default class MessageReceiver
|
|||
if (changed) {
|
||||
log.info('handleBlocked: Block list changed, forcing re-render.');
|
||||
const uniqueIdentifiers = Array.from(new Set(allIdentifiers));
|
||||
window.ConversationController.forceRerender(uniqueIdentifiers);
|
||||
void window.ConversationController.forceRerender(uniqueIdentifiers);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue