Home Reference Source

src/chat/Chat.js

import {List, Map} from 'immutable';
import {BehaviorSubject, combineLatest} from 'rxjs';
import {map} from 'rxjs/operators';

import {EventEmitter} from '../eventemitter';
import {bindMethods} from '../utils/bindMethods';
import {FileUploadStatus} from '../file-sharing/FileUploadStatus';
import {ChatMessageFile} from './ChatMessageFile';
import {ChatMessage} from './ChatMessage';
import {ChatMessageDirection} from './ChatMessageDirection';
import {ChatMessageStatus} from './ChatMessageStatus';
import {ChatMessageType} from './ChatMessageType';
import {ChatType} from './ChatType';
import {ChatParticipantRole} from './ChatParticipantRole';
import {ChatParticipantState} from './ChatParticipantState';
import {AckCounter} from './AckCounter';

/**
 * Provides access to methods for managing a chat room.
 * See {@link ChatManager} to find how Chat objects are obtained
 *
 * # Events
 * - `name`: Emitted every time the name changes
 * - `subject`: Emitted every time the subject changes.
 * - `participants`: Emitted every time the participants Map changes.
 * - `remoteParticipants`: Emitted every time the remoteParticipants Map changes.
 * - `messages` Emitted every time the messages List changes. (deprecated)
 * - `fileMessages` Emitted every time a fileMessages List changes.
 * - `pendingMessages` Emitted every time the pendingMessages List changes.
 * - `firstUnreadMessage` Emitted every time the firstUnreadMessage changes. (deprecated)
 *
 * Some events are deprecated because an observable is also provided. The use of the observable
 * is encouraged and events will be removed at some time in the future.
 */
export class Chat {
	/**
	 * @protected
	 * @return {Chat}
	 */
	static newInstance(chatService, fileManager, user, {id, type, name, subject, participants} = {}) {
		return new Chat(chatService, fileManager, user, id, type, name, subject, participants);
	}

	/**
	 * @private
	 */
	constructor(chatService, fileManager, user, id, type, name, subject, participants) {
		/** @private */
		this.ownAddress = user.getAddress().toLowerCase();

		/** @private */
		this.chatService = chatService;
		/** @private */
		this.fileManager = fileManager;
		/** @private */
		this.id = id;
		/** @private */
		this.type = type;
		/** @private */
		this.name = name;
		/** @private */
		this.subject = subject;
		/** @private */
		this.participants = Map();
		this.participants = this.participants.withMutations((mutableMap) => {
			participants.forEach(p => mutableMap.set(p.jid, p.affiliation));
		});
		/** @private */
		this.remoteParticipants = Map();
		this.remoteParticipants = this.remoteParticipants.withMutations((mutableMap) => {
			participants.filter(p => p.jid !== this.ownAddress)
				.forEach(p => mutableMap.set(p.jid, ChatParticipantState.ACTIVE));
		});
		/** @private */
		this.pendingMessages = new List();
		/** @private */
		this.messagesSubject = new BehaviorSubject(new List());
		/** @private */
		this.fileMessages = new List();
		/** @private */
		this.uploads = new window.Map();
		/** @private */
		this.firstUnreadMessageSubject = new BehaviorSubject();
		/** @private */
		this.lastInvitationTimestamp = null;

		/** @private */
		this.lastRecoveredMessageId = null;
		/** @private */
		this.oldestMessageRecovered = false;
		/** @private */
		this.receivedMarkCounter = new AckCounter();
		this.receivedMarkCounter.setLimit(this.participants.count());
		/** @private */
		this.displayedMarkCounter = new AckCounter();
		this.displayedMarkCounter.setLimit(this.participants.count());
		/** @private */
		this.ownMessageCache = new window.Map();

		/** @type {EventEmitter} */
		this.emitter = new EventEmitter();
		bindMethods(this, [
			'onReceived',
			'onDisplayed',
			'onState',
			'onMessage',
			'onMessageSent',
			'onMucPresence',
			'onMucConfig',
		]);
	}

	/** @protected */
	bindEventListeners() {
		this.chatService.emitter.on(['received', this.id], this.onReceived);
		this.chatService.emitter.on(['displayed', this.id], this.onDisplayed);
		this.chatService.emitter.on(['state', this.id], this.onState);
		this.chatService.emitter.on(['message', this.id], this.onMessage);
		this.chatService.emitter.on(['message:sent', this.id], this.onMessageSent);
		if (this.type === ChatType.GROUP) {
			this.chatService.emitter.on(['mucPresence', this.id], this.onMucPresence);
			this.chatService.emitter.on(['mucConfig', this.id], this.onMucConfig);
		}
	}

	/** @protected */
	unbindEventListeners() {
		this.chatService.emitter.off(['received', this.id], this.onReceived);
		this.chatService.emitter.off(['displayed', this.id], this.onDisplayed);
		this.chatService.emitter.off(['state', this.id], this.onState);
		this.chatService.emitter.off(['message', this.id], this.onMessage);
		this.chatService.emitter.off(['message:sent', this.id], this.onMessageSent);
		if (this.type === ChatType.GROUP) {
			this.chatService.emitter.off(['mucPresence', this.id], this.onMucPresence);
			this.chatService.emitter.off(['mucConfig', this.id], this.onMucConfig);
		}
	}

	/** @private */
	async onMessageSent(event) {
		if (event.received) {
			this.onReceived({
				id: event.received,
				from: this.ownAddress,
			});
			return;
		}
		if (event.displayed) {
			this.onDisplayed({
				id: event.displayed,
				from: this.ownAddress,
			});
			return;
		}
		if (this.type === ChatType.GROUP) {
			this.ownMessageCache.set(event.id, true);
		}
		const type = event.links ? ChatMessageType.FILE : ChatMessageType.TEXT;
		const data = event.links ? new ChatMessageFile(event.links[0].description, event.links[0].url) : event.body;
		const message = ChatMessage.of(this.ownAddress, data, {
			id: event.id,
			type,
			direction: ChatMessageDirection.OUTGOING,
			status: ChatMessageStatus.SENT,
			timestamp: event.timestamp,
			progress: 100,
		});
		this.appendMessage(message);
	}

	/** @private */
	updateMessages(untilId, status) {
		const lastIndex = this.messages.findLastIndex(message => message.getId() === untilId);
		this.messagesSubject.next(this.messages.map((message, i) => i <= lastIndex ? message.with({status}) : message));
	}

	/** @private */
	onReceived(event) {
		if (!this.receivedMarkCounter.hasElement(event.id)) {
			return;
		}
		const lastId = this.receivedMarkCounter.appendAck(event.id, event.from);
		if (!lastId) {
			return;
		}
		this.updateMessages(lastId, ChatMessageStatus.RECEIVED);
		if (this.firstUnreadMessage) {
			this.firstUnreadMessageSubject.next(this.messages.find(message => message.getId() === this.firstUnreadMessage.getId()));
			this.emitter.emit('firstUnreadMessage');
		}
		this.emitter.emit('messages');

		let changed = false;
		this.messages.forEach((message) => {
			if (message.getType() === ChatMessageType.FILE) {
				changed = true;
				const i = this.fileMessages.findIndex(fileMessage => fileMessage.getId() === message.getId());
				this.fileMessages = this.fileMessages.set(i, message);
			}
		});
		if (changed) {
			this.emitter.emit('fileMessages');
		}
	}

	/** @private */
	onDisplayed(event) {
		if (event.from === this.ownAddress) {
			this.updateFirstUnreadMessage(event.id);
		}
		if (this.receivedMarkCounter.hasElement(event.id)) {
			this.receivedMarkCounter.appendAck(event.id, event.from);
		}
		if (!this.displayedMarkCounter.hasElement(event.id)) {
			return;
		}
		const lastId = this.displayedMarkCounter.appendAck(event.id, event.from);
		if (!lastId) {
			return;
		}
		this.updateMessages(lastId, ChatMessageStatus.DISPLAYED);
		if (this.firstUnreadMessage) {
			this.firstUnreadMessageSubject.next(this.messages.find(message => message.getId() === this.firstUnreadMessage.getId()));
			this.emitter.emit('firstUnreadMessage');
		}
		this.emitter.emit('messages');

		let changed = false;
		this.messages.forEach((message) => {
			if (message.getType() === ChatMessageType.FILE) {
				changed = true;
				const i = this.fileMessages.findIndex(fileMessage => fileMessage.getId() === message.getId());
				this.fileMessages = this.fileMessages.set(i, message);
			}
		});
		if (changed) {
			this.emitter.emit('fileMessages');
		}
	}

	/** @private */
	onState(event) {
		if (event.from === this.ownAddress) {
			return;
		}
		this.updateRemoteParticipant(event.from, event.state);
	}

	/** @private */
	async onMessage(event) {
		if (this.type === ChatType.GROUP && this.ownMessageCache.has(event.id)) {
			this.ownMessageCache.delete(event.id);
			return;
		}
		const type = event.links ? ChatMessageType.FILE : ChatMessageType.TEXT;
		const data = event.links ? new ChatMessageFile(event.links[0].description, event.links[0].url) : event.body;
		const direction = event.from === this.ownAddress ? ChatMessageDirection.OUTGOING : ChatMessageDirection.INCOMING;
		const status = direction === ChatMessageDirection.OUTGOING ? ChatMessageStatus.SENT : ChatMessageStatus.RECEIVED;
		const message = ChatMessage.of(event.from, data, {
			id: event.id,
			type,
			direction,
			timestamp: event.timestamp,
			status,
			progress: 100,
		});
		this.appendMessage(message);
		if (!this.firstUnreadMessage) {
			this.firstUnreadMessageSubject.next(this.messages.last());
			this.emitter.emit('firstUnreadMessage');
		}
	}

	/** @private */
	onMucPresence(event) {
		const state = event.affiliation && event.affiliation !== 'none' ?
			this.remoteParticipants.get(event.jid) || ChatParticipantState.ACTIVE : undefined;
		const role = event.affiliation && event.affiliation !== 'none' ? event.affiliation : undefined;
		this.updateRemoteParticipant(event.jid, state);
		this.updateParticipant(event.jid, role);
	}

	/** @private */
	onMucConfig(event) {
		if (typeof event.name === 'string' && this.name !== event.name) {
			this.name = event.name;
			this.emitter.emit('name');
		}
		if (typeof event.subject === 'string' && this.subject !== event.subject) {
			this.subject = event.subject;
			this.emitter.emit('subject');
		}
	}

	/** @private */
	updateRemoteParticipant(id, state) {
		if (!state) {
			this.remoteParticipants = this.remoteParticipants.delete(id);
		} else {
			this.remoteParticipants = this.remoteParticipants.set(id, state);
		}
		this.emitter.emit('remoteParticipants');
	}

	/** @private */
	updateParticipant(id, role) {
		if (!role) {
			this.participants = this.participants.delete(id);
		} else {
			this.participants = this.participants.set(id, role);
		}
		this.receivedMarkCounter.setLimit(this.participants.count());
		this.displayedMarkCounter.setLimit(this.participants.count());
		this.emitter.emit('participants');
	}

	/**
	 * Recovers older messages of this chat. At least the number of messages specified is guaranteed
	 * to be recovered from the server if available. Only if the oldest message is reached a lower
	 * number of messages will be recovered. If that happens, it could be detected using
	 * {@link Chat#isOldestMessageRecovered}.
	 * Recovered messages will be included in the list of messages that is returned using {@link Chat#getMessages}.
	 * and are guaranteed to be accessible when the promise returned by this method is resolved.
	 * @param {number} [minimum=10] The minimum number of messages that will be retrieved if available.
	 * @return {Promise<void>}
	 */
	async fetch(minimum = 10) {
		if (this.oldestMessageRecovered) {
			if (this.type === ChatType.GROUP && this.messages.isEmpty()) {
				const res = await this.chatService.fetchMessages(this.id, 'chat', null, 1);
				if (res.results.length > 0) {
					const message = res.results[0];
					this.lastInvitationTimestamp = message.timestamp;
				}
			}
			return;
		}
		const history = await this.chatService.fetchMessages(this.id, this.type, this.lastRecoveredMessageId, 10 * minimum);
		for (const item of history.results) {
			const composer = item.from;
			if (item.received) {
				this.receivedMarkCounter.prependAck(item.received, composer);
				continue;
			}
			if (item.displayed) {
				this.receivedMarkCounter.prependAck(item.displayed, composer);
				this.displayedMarkCounter.prependAck(item.displayed, composer);
				continue;
			}

			const id = item.id;
			if (this.messages.find(message => id === message.getId())) {
				continue; // Duplicate message
			}
			this.receivedMarkCounter.prepend(id);
			this.displayedMarkCounter.prepend(id);
			const direction = this.ownAddress === composer ? ChatMessageDirection.OUTGOING : ChatMessageDirection.INCOMING;

			const nextMessage = this.messages.first();
			let status;
			if (nextMessage && nextMessage.getStatus() === ChatMessageStatus.DISPLAYED || this.displayedMarkCounter.isOlderAcked()) {
				status = ChatMessageStatus.DISPLAYED;
			} else if (nextMessage && nextMessage.getStatus() === ChatMessageStatus.RECEIVED || this.receivedMarkCounter.isOlderAcked()) {
				status = ChatMessageStatus.RECEIVED;
			} else {
				status = ChatMessageStatus.SENT;
			}

			const timestamp = item.timestamp;
			const type = item.links ? ChatMessageType.FILE : ChatMessageType.TEXT;
			const data = item.links ? new ChatMessageFile(item.links[0].description, item.links[0].url) : item.body;
			const message = ChatMessage.of(composer, data, {id, direction, type, timestamp, status});

			minimum--;
			this.messagesSubject.next(this.messages.unshift(message));
		}
		if (history.complete) {
			this.oldestMessageRecovered = true;
		} else {
			this.lastRecoveredMessageId = history.paging.first;
		}

		if (this.messages.count()) {
			// Mark as received latest message if not done before
			const lastReceivedMessageId = this.receivedMarkCounter.getLastAckedBy(this.ownAddress);
			if (this.messages.last().getId() !== lastReceivedMessageId) {
				await this.chatService.markReceived(this.id, this.messages.last().getId(), this.type);
			}
			this.updateFirstUnreadMessage(this.displayedMarkCounter.getLastAckedBy(this.ownAddress));
		}

		// Request more messages if minimum quantity was not retrieved
		if (minimum > 0) {
			return this.fetch(minimum);
		}
		this.emitter.emit('messages');
	}

	/** @private */
	updateFirstUnreadMessage(lastDisplayedMessageId) {
		const oldFirstUnreadMessage = this.firstUnreadMessage;
		if (!lastDisplayedMessageId) {
			this.firstUnreadMessageSubject.next(this.messages.first());
		} else {
			const lastReadMessageIndex = this.messages.findIndex(message => message.getId() === lastDisplayedMessageId);
			this.firstUnreadMessageSubject.next(this.messages.get(lastReadMessageIndex + 1));
		}
		if (this.firstUnreadMessage !== oldFirstUnreadMessage) {
			this.emitter.emit('firstUnreadMessage');
		}
	}

	/**
	 * Indicates if the oldest message of the chat has already been recovered.
	 * @return {boolean}
	 */
	isOldestMessageRecovered() {
		return this.oldestMessageRecovered;
	}

	/**
	 * Retrieves messages since last message fetching (new messages)
	 *
	 * This is to catch up from the server's MAM
	 *
	 * @return {Promise<void>}
	 */
	async sync() {
		const lastMessage = this.messages.last();
		if (!lastMessage) {
			return this.fetch();
		}
		const since = lastMessage.getTimestamp();
		const results = await this.chatService.sync(this.getId(), this.type, since);
		for (const item of results) {
			if (item.received) {
				this.onReceived({
					id: item.received,
					from: item.from,
				});
				continue;
			}
			if (item.displayed) {
				this.onDisplayed({
					id: item.displayed,
					from: item.from,
				});
				continue;
			}
			if (this.messages.find(message => message.getId() === item.id)) {
				continue;
			}
			const type = item.links ? ChatMessageType.FILE : ChatMessageType.TEXT;
			const data = item.links ? new ChatMessageFile(item.links[0].description, item.links[0].url) : item.body;
			const direction = this.ownAddress === item.from ? ChatMessageDirection.OUTGOING : ChatMessageDirection.INCOMING;
			const message = ChatMessage.of(item.from, data, {
				id: item.id,
				type,
				direction: direction,
				status: ChatMessageStatus.SENT,
				timestamp: item.timestamp,
			});
			this.appendMessage(message);
			if (!this.firstUnreadMessage) {
				this.firstUnreadMessageSubject.next(this.messages.last());
				this.emitter.emit('firstUnreadMessage');
			}
		}
	}

	/**
	 * The ID of this chat. The ID is only locally unique as it is derived from the
	 * remote participant in the case of individual chats and from the room address
	 * for group chats.
	 * @return {string} Unique identifier of the chat
	 */
	getId() {
		return this.id;
	}

	/**
	 * Returns the type of the chat.
	 * @return {ChatType} Type of the chat
	 */
	getType() {
		return this.type;
	}

	/**
	 * Returns the name of the chat. This is only applicable for group chats.
	 * @return {string} Name of the chat
	 */
	getName() {
		return this.name;
	}

	/**
	 * Changes the name of the chat. This is only applicable for group chats.
	 * @param {string} name The new name
	 * @return {Promise}
	 */
	async setName(name) {
		await this.chatService.setRoomName(this.id, name);
		this.onMucConfig({name});
	}

	/**
	 * Returns the subject of the chat. This is only applicable for group chats
	 * @return {string} Subject of the chat
	 */
	getSubject() {
		return this.subject;
	}

	/**
	 * Changes the subject of the chat group. This is only applicable for group chats
	 * @param {string} subject The new subject
	 * @return {Promise}
	 */
	async setSubject(subject) {
		await this.chatService.setRoomSubject(this.id, subject);
		this.onMucConfig({subject});
	}

	/**
	 * Returns the role of the user in the chat. This is only applicable for group chats.
	 * @return {ChatParticipantRole} The role of this user in the chat
	 */
	getRole() {
		return this.getParticipants().get(this.ownAddress);
	}

	/**
	 * Returns a Map with the state of each remote participant.
	 * @return {ImmutableMap<string, ChatParticipantState>}
	 */
	getRemoteParticipants() {
		return this.remoteParticipants;
	}

	/**
	 * Returns a Map with the role of each participant.
	 * @return {ImmutableMap<string, ChatParticipantRole>}
	 */
	getParticipants() {
		return this.participants;
	}

	/**
	 * Adds a new participant to the chat.
	 * @param {string} participant The address of the user that wants to be added
	 * @return {Promise}
	 */
	async addParticipant(participant) {
		participant = participant.toLowerCase();
		await this.chatService.invite(this.id, participant);
		this.updateRemoteParticipant(participant, ChatParticipantState.ACTIVE);
		this.updateParticipant(participant, ChatParticipantRole.MEMBER);
	}

	/**
	 * Expels a participant from the chat.
	 * @param {string} participant The address of the user that will be expelled
	 * @return {Promise}
	 */
	expelParticipant(participant) {
		return this.chatService.expel(this.id, participant);
	}

	/**
	 * Changes the role of a chat participant.
	 * @param {string} participant The address of the user whose role wants to be changed
	 * @param {ChatParticipantRole} role The new role of this participant
	 * @return {Promise}
	 */
	setParticipantRole(participant, role) {
		return this.chatService.setParticipantRole(this.id, participant, role);
	}

	/**
	 * Leaves the group chat. Only group chats can be deleted and only members can leave them.
	 * Admins need to stop being an admin before.
	 * @return {Promise}
	 */
	async leave() {
		if (this.getRole() === ChatParticipantRole.ADMIN) {
			throw Error('Admins cannot leave room');
		}
		return this.chatService.leave(this.id);
	}

	/**
	 * Removes the group chat. Only group chats can be deleted and only admins can delete them.
	 * @return {Promise}
	 */
	async delete() {
		if (this.getType() !== ChatType.GROUP) {
			throw Error('Only group chats can be deleted');
		}
		await this.chatService.destroy(this.getId());
	}

	/**
	 * Removes the chat from the list of chats. Only individual chats can be archived.
	 * Group chats need to be leaved or deleted instead.
	 * @return {Promise}
	 */
	async archive() {
		if (this.getType() !== ChatType.INDIVIDUAL) {
			throw Error('Only individual chats can be archived');
		}
		await this.chatService.deleteIndividualChat(this.getId());
	}

	/**
	 * Updates the state of the local participant in the chat.
	 * @param {ChatParticipantState} state the new state of the local particpant
	 * @return {Promise}
	 */
	setState(state) {
		return this.chatService.setState(this.getId(), this.type, state);
	}

	/** @private */
	appendMessage(message) {
		this.receivedMarkCounter.append(message.getId());
		this.displayedMarkCounter.append(message.getId());
		this.messagesSubject.next(this.messages.push(message));
		this.emitter.emit('messages', message);
		if (message.getType() === ChatMessageType.FILE) {
			this.fileMessages = this.fileMessages.push(message);
			this.emitter.emit('fileMessages', message);
		}
	}

	/**
	 * Sends a new text message.
	 * @param {string} text The text of the new message
	 * @return {Promise} Resolved when the message is received by the server
	 */
	async sendText(text) {
		const pendingMsg = ChatMessage.of(this.ownAddress, text, {
			direction: ChatMessageDirection.OUTGOING,
			timestamp: Date.now(),
		});
		this.pendingMessages = this.pendingMessages.push(pendingMsg);
		this.emitter.emit('pendingMessages');
		try {
			await this.chatService.sendMessage(this.id, this.type, text);
		} catch (error) {
			const message = pendingMsg.with({status: ChatMessageStatus.ERROR});
			this.pendingMessages = this.pendingMessages.map(msg => msg === pendingMsg ? message : msg);
			this.emitter.emit('pendingMessages');
			throw error;
		}
		this.pendingMessages = this.pendingMessages.filter(msg => msg !== pendingMsg);
		this.emitter.emit('pendingMessages');
		await this.resetUnreadMessageCount();
	}

	/**
	 * Sends a new file message.
	 * @param {File} file A file that wants to be sent to this chat
	 * @return {Promise} Resolved when the message is received by the server
	 */
	async sendFile(file) {
		const fileUploadManager = await this.fileManager.getFileUploadManager();
		const fileUploadId = await fileUploadManager.create(file);
		const errorPromise = new Promise((resolve, reject) => fileUploadManager.emitter.once([fileUploadId, 'error'], reject));
		const startPromise = new Promise(resolve => fileUploadManager.emitter.once([fileUploadId, 'start'], resolve));
		try {
			await Promise.race([errorPromise, startPromise]);
		} catch (fileUpload) {
			return fileUpload;
		}

		let pendingMsg = ChatMessage.of(this.ownAddress, new ChatMessageFile(file.name), {
			direction: ChatMessageDirection.OUTGOING,
			timestamp: Date.now(),
			type: ChatMessageType.FILE,
		});
		this.pendingMessages = this.pendingMessages.push(pendingMsg);
		this.uploads.set(pendingMsg, fileUploadId);
		this.emitter.emit('pendingMessages', pendingMsg, null);

		let updatePendingMessage = () => {
			const fileUpload = fileUploadManager.getFileUploads().get(fileUploadId);
			const progress = fileUpload.getProgress();
			const data = new ChatMessageFile(fileUpload.getName(), fileUpload.getUrl());
			const nextMessage = pendingMsg.with({progress, data});
			this.pendingMessages = this.pendingMessages.map(msg => msg === pendingMsg ? nextMessage : msg);
			this.uploads.delete(pendingMsg);
			this.uploads.set(nextMessage, fileUploadId);
			this.emitter.emit('pendingMessages', nextMessage, pendingMsg);
			pendingMsg = nextMessage;
		};
		fileUploadManager.emitter.on([fileUploadId, 'progress'], updatePendingMessage);
		let completeFileUpload;
		try {
			this.uploads.delete(pendingMsg);
			const fileUpload = await fileUploadManager.waitForEnd(fileUploadId);
			if (fileUpload.getStatus() !== FileUploadStatus.COMPLETED) {
				this.pendingMessages = this.pendingMessages.filter(msg => msg !== pendingMsg);
				this.emitter.emit('pendingMessages');
				return fileUpload;
			}
			this.chatService.sendFile(this.id, this.type, fileUpload.getUrl(), fileUpload.getName());
		} catch (fileUpload) {
			const newMessage = pendingMsg.with({status: ChatMessageStatus.ERROR});
			this.pendingMessages = this.pendingMessages.map(msg => msg === pendingMsg ? newMessage : msg);
			this.emitter.emit('pendingMessages');
			return fileUpload;
		} finally {
			const fileUpload = fileUploadManager.getFileUploads().get(fileUploadId);
			completeFileUpload = fileUpload;
			fileUploadManager.emitter.off([fileUploadId, 'progress'], updatePendingMessage);
			fileUploadManager.remove(fileUpload);
		}
		this.pendingMessages = this.pendingMessages.filter(msg => msg !== pendingMsg);
		this.emitter.emit('pendingMessages');
		await this.resetUnreadMessageCount();
		return completeFileUpload;
	}

	/**
	 * Cancels a pending file message. This is only applicable for file messages.
	 * @param {ChatMessage} message The chat message that wants to be cancelled.
	 * @return {Promise<void>}
	 */
	async cancelPendingMessage(message) {
		if (!this.pendingMessages.includes(message) || message.getType() !== ChatMessageType.FILE) {
			throw Error('Invalid message');
		}
		const fileUploadManager = await this.fileManager.getFileUploadManager();
		const fileUploadId = this.uploads.get(message);
		const fileUpload = fileUploadManager.getFileUploads().get(fileUploadId);
		await fileUploadManager.abort(fileUpload);
	}

	/**
	 * Returns the timestamp of the last sent or received message including group invitations
	 * @return {number}
	 */
	getLastMessageTimestamp() {
		if (this.getMessages().size > 0) {
			return this.getMessages().last().getTimestamp();
		}
		return this.lastInvitationTimestamp;
	}

	/**
	 * Returns the list of messages that have not been received by the server yet.
	 * @return {ImmutableList<ChatMessage>}
	 */
	getPendingMessages() {
		return this.pendingMessages;
	}

	/**
	 * Returns the list of chat messages.
	 * @return {ImmutableList<ChatMessage>}
	 */
	getMessages$() {
		return this.messagesSubject.asObservable();
	}

	/**
	 * Returns the list of chat messages.
	 * @return {ImmutableList<ChatMessage>}
	 */
	getMessages() {
		return this.messagesSubject.value;
	}

	/**
	 * The list of chat messages.
	 * @type {ImmutableList<ChatMessage>}
	 */
	get messages() {
		return this.messagesSubject.value;
	}

	/**
	 * Returns the list of chat messages with file type.
	 * @return {ImmutableList<ChatMessage>}
	 */
	getFileMessages() {
		return this.fileMessages;
	}

	/**
	 * Returns an Observable of the first unread message or undefined if every message is already read.
	 * @return {Observable<ChatMessage>}
	 */
	getFirstUnreadMessage$() {
		return this.firstUnreadMessageSubject.asObservable();
	}

	/**
	 * Returns the first unread message or undefined if every message is already read.
	 * @return {ChatMessage}
	 */
	getFirstUnreadMessage() {
		return this.firstUnreadMessage;
	}

	/**
	 * The the first unread message or undefined if every message is already read.
	 * @type {ChatMessage}
	 */
	get firstUnreadMessage() {
		return this.firstUnreadMessageSubject.value;
	}

	/**
	 * Returns the count of unread messages.
	 * @return {number}
	 */
	getUnreadMessageCount() {
		return this.messages.skipUntil(message => message === this.firstUnreadMessage).count();
	}

	/**
	 * Returns an observable of the count of unread messages.
	 * @return {Observable<number>}
	 */
	getUnreadMessageCount$() {
		return combineLatest(this.messagesSubject, this.firstUnreadMessageSubject).pipe(
			map(([messages, firstUnreadMessage]) => messages.skipUntil(message => message === firstUnreadMessage).count()),
		);
	}

	/**
	 * Resets the count of unread messages.
	 * @return {Promise}
	 */
	async resetUnreadMessageCount() {
		const lastMessage = this.messages.last();
		if (!lastMessage || lastMessage.getStatus() === ChatMessageStatus.DISPLAYED) {
			return;
		}
		await this.chatService.markDisplayed(this.id, lastMessage.getId(), this.type);
		if (this.type === ChatType.GROUP) {
			await new Promise((resolve) => {
				const onDisplayed = (event) => {
					if (event.id === lastMessage.getId() && event.from === this.ownAddress) {
						this.chatService.emitter.off(['displayed', this.id], onDisplayed);
						resolve();
					}
				};
				this.chatService.emitter.on(['displayed', this.id], onDisplayed);
			});
		}
	}
}