import {ExSocketInterface} from "./Websocket/ExSocketInterface"; import {apiClientRepository} from "../Services/ApiClientRepository"; import { BatchToPusherMessage, CharacterLayerMessage, GroupLeftZoneMessage, GroupUpdateMessage, GroupUpdateZoneMessage, PointMessage, PositionMessage, UserJoinedMessage, UserJoinedZoneMessage, UserLeftZoneMessage, UserMovedMessage, ZoneMessage } from "../Messages/generated/messages_pb"; import * as messages_pb from "../Messages/generated/messages_pb"; import {ClientReadableStream} from "grpc"; import {PositionDispatcher} from "_Model/PositionDispatcher"; import {socketManager} from "../Services/SocketManager"; import {ProtobufUtils} from "_Model/Websocket/ProtobufUtils"; import Debug from "debug"; const debug = Debug("zone"); export interface ZoneEventListener { onUserEnters(user: UserDescriptor, listener: ExSocketInterface): void; onUserMoves(user: UserDescriptor, listener: ExSocketInterface): void; onUserLeaves(userId: number, listener: ExSocketInterface): void; onGroupEnters(group: GroupDescriptor, listener: ExSocketInterface): void; onGroupMoves(group: GroupDescriptor, listener: ExSocketInterface): void; onGroupLeaves(groupId: number, listener: ExSocketInterface): void; } /*export type EntersCallback = (thing: Movable, listener: User) => void; export type MovesCallback = (thing: Movable, position: PositionInterface, listener: User) => void; export type LeavesCallback = (thing: Movable, listener: User) => void;*/ export class UserDescriptor { private constructor(public readonly userId: number, private name: string, private characterLayers: CharacterLayerMessage[], private position: PositionMessage) { if (!Number.isInteger(this.userId)) { throw new Error('UserDescriptor.userId is not an integer: '+this.userId); } } public static createFromUserJoinedZoneMessage(message: UserJoinedZoneMessage): UserDescriptor { const position = message.getPosition(); if (position === undefined) { throw new Error('Missing position'); } return new UserDescriptor(message.getUserid(), message.getName(), message.getCharacterlayersList(), position); } public update(userMovedMessage: UserMovedMessage) { const position = userMovedMessage.getPosition(); if (position === undefined) { throw new Error('Missing position'); } this.position = position; } public toUserJoinedMessage(): UserJoinedMessage { const userJoinedMessage = new UserJoinedMessage(); userJoinedMessage.setUserid(this.userId); userJoinedMessage.setName(this.name); userJoinedMessage.setCharacterlayersList(this.characterLayers); userJoinedMessage.setPosition(this.position); return userJoinedMessage; } public toUserMovedMessage(): UserMovedMessage { const userMovedMessage = new UserMovedMessage(); userMovedMessage.setUserid(this.userId); userMovedMessage.setPosition(this.position); return userMovedMessage; } } export class GroupDescriptor { private constructor(public readonly groupId: number, private groupSize: number, private position: PointMessage) { } public static createFromGroupUpdateZoneMessage(message: GroupUpdateZoneMessage): GroupDescriptor { const position = message.getPosition(); if (position === undefined) { throw new Error('Missing position'); } return new GroupDescriptor(message.getGroupid(), message.getGroupsize(), position); } public update(groupDescriptor: GroupDescriptor) { this.groupSize = groupDescriptor.groupSize; this.position = groupDescriptor.position; } public toGroupUpdateMessage(): GroupUpdateMessage { const groupUpdateMessage = new GroupUpdateMessage(); if (!Number.isInteger(this.groupId)) { throw new Error('GroupDescriptor.groupId is not an integer: '+this.groupId); } groupUpdateMessage.setGroupid(this.groupId); groupUpdateMessage.setGroupsize(this.groupSize); groupUpdateMessage.setPosition(this.position); return groupUpdateMessage; } } interface ZoneDescriptor { x: number, y: number } export class Zone { //private things: Set = new Set(); private users: Map = new Map(); private groups: Map = new Map(); private listeners: Set = new Set(); private backConnection!: ClientReadableStream; private isClosing: boolean = false; constructor(private positionDispatcher: PositionDispatcher, private socketListener: ZoneEventListener, public readonly x: number, public readonly y: number, private onBackFailure: (e: Error|null, zone: Zone) => void) { } /** * Creates a connection to the back server to track the users. */ public async init(): Promise { debug('Opening connection to zone %d, %d on back server', this.x, this.y); const apiClient = await apiClientRepository.getClient(this.positionDispatcher.roomId); const zoneMessage = new ZoneMessage(); zoneMessage.setRoomid(this.positionDispatcher.roomId); zoneMessage.setX(this.x); zoneMessage.setY(this.y); this.backConnection = apiClient.listenZone(zoneMessage); this.backConnection.on('data', (batch: BatchToPusherMessage) => { for (const message of batch.getPayloadList()) { if (message.hasUserjoinedzonemessage()) { const userJoinedZoneMessage = message.getUserjoinedzonemessage() as UserJoinedZoneMessage; const userDescriptor = UserDescriptor.createFromUserJoinedZoneMessage(userJoinedZoneMessage); this.users.set(userJoinedZoneMessage.getUserid(), userDescriptor); const fromZone = userJoinedZoneMessage.getFromzone(); this.notifyUserEnter(userDescriptor, fromZone?.toObject()); } else if (message.hasGroupupdatezonemessage()) { const groupUpdateZoneMessage = message.getGroupupdatezonemessage() as GroupUpdateZoneMessage; const groupDescriptor = GroupDescriptor.createFromGroupUpdateZoneMessage(groupUpdateZoneMessage); // Do we have it already? const groupId = groupUpdateZoneMessage.getGroupid(); const oldGroupDescriptor = this.groups.get(groupId); if (oldGroupDescriptor !== undefined) { oldGroupDescriptor.update(groupDescriptor); this.notifyGroupMove(groupDescriptor); } else { this.groups.set(groupId, groupDescriptor); const fromZone = groupUpdateZoneMessage.getFromzone(); this.notifyGroupEnter(groupDescriptor, fromZone?.toObject()); } } else if (message.hasUserleftzonemessage()) { const userLeftMessage = message.getUserleftzonemessage() as UserLeftZoneMessage; this.users.delete(userLeftMessage.getUserid()); this.notifyUserLeft(userLeftMessage.getUserid(), userLeftMessage.getTozone()?.toObject()); } else if (message.hasGroupleftzonemessage()) { const groupLeftMessage = message.getGroupleftzonemessage() as GroupLeftZoneMessage; this.groups.delete(groupLeftMessage.getGroupid()); this.notifyGroupLeft(groupLeftMessage.getGroupid(), groupLeftMessage.getTozone()?.toObject()); } else if (message.hasUsermovedmessage()) { const userMovedMessage = message.getUsermovedmessage() as UserMovedMessage; const userId = userMovedMessage.getUserid(); const userDescriptor = this.users.get(userId); if (userDescriptor === undefined) { console.error('Unexpected move message received for user "'+userId+'"'); return; } userDescriptor.update(userMovedMessage); this.notifyUserMove(userDescriptor); } else { throw new Error('Unexpected message'); } } }); this.backConnection.on('error', (e) => { if (!this.isClosing) { debug('Error on back connection') this.close(); this.onBackFailure(e, this); } }); this.backConnection.on('close', () => { if (!this.isClosing) { debug('Close on back connection') this.close(); this.onBackFailure(null, this); } }); } public close(): void { debug('Closing connection to zone %d, %d on back server', this.x, this.y); this.isClosing = true; this.backConnection.cancel(); } public hasListeners(): boolean { return this.listeners.size !== 0; } /** * Notify listeners of this zone that this user entered */ private notifyUserEnter(user: UserDescriptor, oldZone: ZoneDescriptor|undefined) { for (const listener of this.listeners) { if (listener.userId === user.userId) { continue; } if (oldZone === undefined || !this.isListeningZone(listener, oldZone.x, oldZone.y)) { this.socketListener.onUserEnters(user, listener); } else { this.socketListener.onUserMoves(user, listener); } } } /** * Notify listeners of this zone that this group entered */ private notifyGroupEnter(group: GroupDescriptor, oldZone: ZoneDescriptor|undefined) { for (const listener of this.listeners) { if (oldZone === undefined || !this.isListeningZone(listener, oldZone.x, oldZone.y)) { this.socketListener.onGroupEnters(group, listener); } else { this.socketListener.onGroupMoves(group, listener); } } } /** * Notify listeners of this zone that this user left */ private notifyUserLeft(userId: number, newZone: ZoneDescriptor|undefined) { for (const listener of this.listeners) { if (listener.userId === userId) { continue; } if (newZone === undefined || !this.isListeningZone(listener, newZone.x, newZone.y)) { this.socketListener.onUserLeaves(userId, listener); } else { // Do not send a signal. The move event will be triggered when joining the new room. } } } /** * Notify listeners of this zone that this group left */ private notifyGroupLeft(groupId: number, newZone: ZoneDescriptor|undefined) { for (const listener of this.listeners) { if (listener.groupId === groupId) { continue; } if (newZone === undefined || !this.isListeningZone(listener, newZone.x, newZone.y)) { this.socketListener.onGroupLeaves(groupId, listener); } else { // Do not send a signal. The move event will be triggered when joining the new room. } } } private isListeningZone(socket: ExSocketInterface, x: number, y: number): boolean { // TODO: improve efficiency by not doing a full scan of listened zones. for (const zone of socket.listenedZones) { if (zone.x === x && zone.y === y) { return true; } } return false; } private notifyGroupMove(groupDescriptor: GroupDescriptor) { for (const listener of this.listeners) { this.socketListener.onGroupMoves(groupDescriptor, listener); } } private notifyUserMove(userDescriptor: UserDescriptor) { for (const listener of this.listeners) { if (listener.userId === userDescriptor.userId) { continue; } this.socketListener.onUserMoves(userDescriptor, listener); } } public startListening(listener: ExSocketInterface): void { for (const [userId, user] of this.users.entries()) { if (userId !== listener.userId) { this.socketListener.onUserEnters(user, listener); } } for (const [groupId, group] of this.groups.entries()) { this.socketListener.onGroupEnters(group, listener); } this.listeners.add(listener); listener.listenedZones.add(this); } public stopListening(listener: ExSocketInterface): void { for (const [userId, user] of this.users.entries()) { if (userId !== listener.userId) { this.socketListener.onUserLeaves(userId, listener); } } for (const [groupId, group] of this.groups.entries()) { this.socketListener.onGroupLeaves(groupId, listener); } this.listeners.delete(listener); listener.listenedZones.delete(this); } }