|
|
|
@ -6,6 +6,7 @@ import {
|
|
|
|
|
} from './MessageQueueInterface';
|
|
|
|
|
import {
|
|
|
|
|
ChatMessage,
|
|
|
|
|
ClosedGroupNewMessage,
|
|
|
|
|
ContentMessage,
|
|
|
|
|
DataMessage,
|
|
|
|
|
ExpirationTimerUpdateMessage,
|
|
|
|
@ -40,8 +41,7 @@ export class MessageQueue implements MessageQueueInterface {
|
|
|
|
|
) {
|
|
|
|
|
throw new Error('SyncMessage needs to be sent with sendSyncMessage');
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
await this.sendMessageToDevices([user], message);
|
|
|
|
|
await this.process(user, message, sentCb);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public async send(
|
|
|
|
@ -55,7 +55,7 @@ export class MessageQueue implements MessageQueueInterface {
|
|
|
|
|
) {
|
|
|
|
|
throw new Error('SyncMessage needs to be sent with sendSyncMessage');
|
|
|
|
|
}
|
|
|
|
|
await this.sendMessageToDevices([device], message, sentCb);
|
|
|
|
|
await this.process(device, message, sentCb);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
@ -136,7 +136,7 @@ export class MessageQueue implements MessageQueueInterface {
|
|
|
|
|
throw new Error('ourNumber is not set');
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
await this.sendMessageToDevices([PubKey.cast(ourPubKey)], message, sentCb);
|
|
|
|
|
await this.process(PubKey.cast(ourPubKey), message, sentCb);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public async processPending(device: PubKey) {
|
|
|
|
@ -172,18 +172,6 @@ export class MessageQueue implements MessageQueueInterface {
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public async sendMessageToDevices(
|
|
|
|
|
devices: Array<PubKey>,
|
|
|
|
|
message: ContentMessage,
|
|
|
|
|
sentCb?: (message: RawMessage) => Promise<void>
|
|
|
|
|
) {
|
|
|
|
|
const promises = devices.map(async device => {
|
|
|
|
|
await this.process(device, message, sentCb);
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
return Promise.all(promises);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private async processAllPending() {
|
|
|
|
|
const devices = await this.pendingMessageCache.getDevices();
|
|
|
|
|
const promises = devices.map(async device => this.processPending(device));
|
|
|
|
@ -191,6 +179,9 @@ export class MessageQueue implements MessageQueueInterface {
|
|
|
|
|
return Promise.all(promises);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* This method should not be called directly. Only through sendToPubKey.
|
|
|
|
|
*/
|
|
|
|
|
private async process(
|
|
|
|
|
device: PubKey,
|
|
|
|
|
message: ContentMessage,
|
|
|
|
@ -199,9 +190,11 @@ export class MessageQueue implements MessageQueueInterface {
|
|
|
|
|
// Don't send to ourselves
|
|
|
|
|
const currentDevice = await UserUtils.getCurrentDevicePubKey();
|
|
|
|
|
if (currentDevice && device.isEqual(currentDevice)) {
|
|
|
|
|
// We allow a message for ourselve only if it's a ConfigurationMessage or a message with a syncTarget set
|
|
|
|
|
// We allow a message for ourselve only if it's a ConfigurationMessage, a ClosedGroupNewMessage,
|
|
|
|
|
// or a message with a syncTarget set.
|
|
|
|
|
if (
|
|
|
|
|
message instanceof ConfigurationMessage ||
|
|
|
|
|
message instanceof ClosedGroupNewMessage ||
|
|
|
|
|
(message as any).syncTarget?.length > 0
|
|
|
|
|
) {
|
|
|
|
|
window.log.warn('Processing sync message');
|
|
|
|
|