use timestamp offset in for messages storage requests (#1892)
* add using timestamp offset from network for sending part1 * remove MessageController as we now rely on the database * fix tests for message sending overriding timestamppull/1921/head
Audric Ackermann
4 years ago
committed by
GitHub
parent
ab75f945ff
commit
b45109985c
@ -1,75 +0,0 @@
|
||||
// You can see MessageController for in memory registered messages.
|
||||
// Ee register messages to it everytime we send one, so that when an event happens we can find which message it was based on this id.
|
||||
|
||||
import { MessageModel } from '../../models/message';
|
||||
|
||||
type MessageControllerEntry = {
|
||||
message: MessageModel;
|
||||
timestamp: number;
|
||||
};
|
||||
let messageControllerInstance: MessageController | null;
|
||||
|
||||
export const getMessageController = () => {
|
||||
if (messageControllerInstance) {
|
||||
return messageControllerInstance;
|
||||
}
|
||||
messageControllerInstance = new MessageController();
|
||||
return messageControllerInstance;
|
||||
};
|
||||
|
||||
// It's not only data from the db which is stored on the MessageController entries, we could fetch this again. What we cannot fetch from the db and which is stored here is all listeners a particular messages is linked to for instance. We will be able to get rid of this once we don't use backbone models at all
|
||||
export class MessageController {
|
||||
private readonly messageLookup: Map<string, MessageControllerEntry>;
|
||||
|
||||
/**
|
||||
* Not to be used directly. Instead call getMessageController()
|
||||
*/
|
||||
constructor() {
|
||||
this.messageLookup = new Map();
|
||||
// cleanup every hour the cache
|
||||
setInterval(this.cleanup, 3600 * 1000);
|
||||
}
|
||||
|
||||
public register(id: string, message: MessageModel) {
|
||||
if (!(message instanceof MessageModel)) {
|
||||
throw new Error('Only MessageModels can be registered to the MessageController.');
|
||||
}
|
||||
const existing = this.messageLookup.get(id);
|
||||
if (existing) {
|
||||
this.messageLookup.set(id, {
|
||||
message: existing.message,
|
||||
timestamp: Date.now(),
|
||||
});
|
||||
return existing.message;
|
||||
}
|
||||
|
||||
this.messageLookup.set(id, {
|
||||
message,
|
||||
timestamp: Date.now(),
|
||||
});
|
||||
|
||||
return message;
|
||||
}
|
||||
|
||||
public unregister(id: string) {
|
||||
this.messageLookup.delete(id);
|
||||
}
|
||||
|
||||
public cleanup() {
|
||||
window?.log?.warn('Cleaning up MessageController singleton oldest messages...');
|
||||
const now = Date.now();
|
||||
|
||||
(this.messageLookup || []).forEach(messageEntry => {
|
||||
const { message, timestamp } = messageEntry;
|
||||
const conversation = message.getConversation();
|
||||
|
||||
if (now - timestamp > 5 * 60 * 1000 && !conversation) {
|
||||
this.unregister(message.id);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public get(identifier: string) {
|
||||
return this.messageLookup.get(identifier);
|
||||
}
|
||||
}
|
||||
@ -1,4 +1,3 @@
|
||||
import * as Outgoing from './outgoing';
|
||||
import { getMessageController } from './MessageController';
|
||||
|
||||
export { Outgoing, getMessageController };
|
||||
export { Outgoing };
|
||||
|
||||
@ -1,80 +0,0 @@
|
||||
import _ from 'lodash';
|
||||
import { storeOnNode } from '../snode_api/SNodeAPI';
|
||||
import { getSwarmFor } from '../snode_api/snodePool';
|
||||
import { firstTrue } from '../utils/Promise';
|
||||
|
||||
const DEFAULT_CONNECTIONS = 3;
|
||||
|
||||
/**
|
||||
* Refactor note: We should really clean this up ... it's very messy
|
||||
*
|
||||
* We need to split it into 2 sends:
|
||||
* - Snodes
|
||||
* - Open Groups
|
||||
*
|
||||
* Mikunj:
|
||||
* Temporarily i've made it so `MessageSender` handles open group sends and calls this function for regular sends.
|
||||
*/
|
||||
|
||||
export async function sendMessage(
|
||||
pubKey: string,
|
||||
data: Uint8Array,
|
||||
messageTimeStamp: number,
|
||||
ttl: number,
|
||||
options: {
|
||||
isPublic?: boolean;
|
||||
} = {}
|
||||
): Promise<void> {
|
||||
const { isPublic = false } = options;
|
||||
|
||||
if (isPublic) {
|
||||
window?.log?.warn('this sendMessage() should not be called anymore with an open group message');
|
||||
return;
|
||||
}
|
||||
|
||||
const data64 = window.dcodeIO.ByteBuffer.wrap(data).toString('base64');
|
||||
|
||||
// Using timestamp as a unique identifier
|
||||
const swarm = await getSwarmFor(pubKey);
|
||||
|
||||
// send parameters
|
||||
const params = {
|
||||
pubKey,
|
||||
ttl: `${ttl}`,
|
||||
timestamp: `${messageTimeStamp}`,
|
||||
data: data64,
|
||||
};
|
||||
|
||||
const usedNodes = _.slice(swarm, 0, DEFAULT_CONNECTIONS);
|
||||
|
||||
const promises = usedNodes.map(async usedNode => {
|
||||
// TODO: Revert back to using snode address instead of IP
|
||||
// No pRetry here as if this is a bad path it will be handled and retried in lokiOnionFetch.
|
||||
// the only case we could care about a retry would be when the usedNode is not correct,
|
||||
// but considering we trigger this request with a few snode in //, this should be fine.
|
||||
const successfulSend = await storeOnNode(usedNode, params);
|
||||
if (successfulSend) {
|
||||
return usedNode;
|
||||
}
|
||||
// should we mark snode as bad if it can't store our message?
|
||||
return undefined;
|
||||
});
|
||||
|
||||
let snode;
|
||||
try {
|
||||
snode = await firstTrue(promises);
|
||||
} catch (e) {
|
||||
const snodeStr = snode ? `${snode.ip}:${snode.port}` : 'null';
|
||||
window?.log?.warn(
|
||||
`loki_message:::sendMessage - ${e.code} ${e.message} to ${pubKey} via snode:${snodeStr}`
|
||||
);
|
||||
throw e;
|
||||
}
|
||||
if (!usedNodes || usedNodes.length === 0) {
|
||||
throw new window.textsecure.EmptySwarmError(pubKey, 'Ran out of swarm nodes to query');
|
||||
}
|
||||
|
||||
window?.log?.info(
|
||||
`loki_message:::sendMessage - Successfully stored message to ${pubKey} via ${snode.ip}:${snode.port}`
|
||||
);
|
||||
}
|
||||
@ -1,7 +1,6 @@
|
||||
// TS 3.8 supports export * as X from 'Y'
|
||||
import * as MessageSender from './MessageSender';
|
||||
import * as LokiMessageApi from './LokiMessageApi';
|
||||
export { MessageSender, LokiMessageApi };
|
||||
export { MessageSender };
|
||||
|
||||
export * from './PendingMessageCache';
|
||||
export * from './MessageQueue';
|
||||
|
||||
Loading…
Reference in New Issue