Kraut.World/server
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'develop' of github.com:thecodingmachine/workadventure into GlobalMessageToWorld

Browse source
This commit is contained in:
GRL 2021-07-22 16:26:01 +02:00
commit 68ff1b9e17
81 changed files with 3001 additions and 1073 deletions
Download patch file Download diff file Expand all files Collapse all files

3
pusher/src/Controller/IoSocketController.ts
View file

@ -17,6 +17,7 @@ import {
ServerToClientMessage,
CompanionMessage,
EmotePromptMessage,
VariableMessage,
} from "../Messages/generated/messages_pb";
import { UserMovesMessage } from "../Messages/generated/messages_pb";
import { TemplatedApp } from "uWebSockets.js";
@ -357,6 +358,8 @@ export class IoSocketController {
socketManager.handleSilentMessage(client, message.getSilentmessage() as SilentMessage);
} else if (message.hasItemeventmessage()) {
socketManager.handleItemEvent(client, message.getItemeventmessage() as ItemEventMessage);
} else if (message.hasVariablemessage()) {
socketManager.handleVariableEvent(client, message.getVariablemessage() as VariableMessage);
} else if (message.hasWebrtcsignaltoservermessage()) {
socketManager.emitVideo(
client,

104
pusher/src/Model/PusherRoom.ts
View file

@ -2,7 +2,29 @@ import { ExSocketInterface } from "_Model/Websocket/ExSocketInterface";
import { PositionDispatcher } from "./PositionDispatcher";
import { ViewportInterface } from "_Model/Websocket/ViewportMessage";
import { arrayIntersect } from "../Services/ArrayHelper";
import { ZoneEventListener } from "_Model/Zone";
import { GroupDescriptor, UserDescriptor, ZoneEventListener } from "_Model/Zone";
import { apiClientRepository } from "../Services/ApiClientRepository";
import {
BatchToPusherMessage,
BatchToPusherRoomMessage,
EmoteEventMessage,
ErrorMessage,
GroupLeftZoneMessage,
GroupUpdateZoneMessage,
RoomMessage,
SubMessage,
UserJoinedZoneMessage,
UserLeftZoneMessage,
UserMovedMessage,
VariableMessage,
VariableWithTagMessage,
ZoneMessage,
} from "../Messages/generated/messages_pb";
import Debug from "debug";
import { ClientReadableStream } from "grpc";
import { ExAdminSocketInterface } from "_Model/Websocket/ExAdminSocketInterface";
const debug = Debug("room");
export enum GameRoomPolicyTypes {
ANONYMOUS_POLICY = 1,
@ -15,6 +37,10 @@ export class PusherRoom {
public tags: string[];
public policyType: GameRoomPolicyTypes;
private versionNumber: number = 1;
private backConnection!: ClientReadableStream<BatchToPusherRoomMessage>;
private isClosing: boolean = false;
private listeners: Set<ExSocketInterface> = new Set<ExSocketInterface>();
//public readonly variables = new Map<string, string>();
constructor(public readonly roomUrl: string, private socketListener: ZoneEventListener) {
this.tags = [];
@ -28,8 +54,13 @@ export class PusherRoom {
this.positionNotifier.setViewport(socket, viewport);
}
public join(socket: ExSocketInterface) {
this.listeners.add(socket);
}
public leave(socket: ExSocketInterface) {
this.positionNotifier.removeViewport(socket);
this.listeners.delete(socket);
}
public canAccess(userTags: string[]): boolean {
@ -48,4 +79,75 @@ export class PusherRoom {
return false;
}
}
/**
* Creates a connection to the back server to track global messages relative to this room (like variable changes).
*/
public async init(): Promise<void> {
debug("Opening connection to room %s on back server", this.roomUrl);
const apiClient = await apiClientRepository.getClient(this.roomUrl);
const roomMessage = new RoomMessage();
roomMessage.setRoomid(this.roomUrl);
this.backConnection = apiClient.listenRoom(roomMessage);
this.backConnection.on("data", (batch: BatchToPusherRoomMessage) => {
for (const message of batch.getPayloadList()) {
if (message.hasVariablemessage()) {
const variableMessage = message.getVariablemessage() as VariableWithTagMessage;
const readableBy = variableMessage.getReadableby();
// We need to store all variables to dispatch variables later to the listeners
//this.variables.set(variableMessage.getName(), variableMessage.getValue(), readableBy);
// Let's dispatch this variable to all the listeners
for (const listener of this.listeners) {
const subMessage = new SubMessage();
if (!readableBy || listener.tags.includes(readableBy)) {
subMessage.setVariablemessage(variableMessage);
}
listener.emitInBatch(subMessage);
}
} else if (message.hasErrormessage()) {
const errorMessage = message.getErrormessage() as ErrorMessage;
// Let's dispatch this error to all the listeners
for (const listener of this.listeners) {
const subMessage = new SubMessage();
subMessage.setErrormessage(errorMessage);
listener.emitInBatch(subMessage);
}
} else {
throw new Error("Unexpected message");
}
}
});
this.backConnection.on("error", (e) => {
if (!this.isClosing) {
debug("Error on back connection");
this.close();
// Let's close all connections linked to that room
for (const listener of this.listeners) {
listener.disconnecting = true;
listener.end(1011, "Connection error between pusher and back server");
}
}
});
this.backConnection.on("close", () => {
if (!this.isClosing) {
debug("Close on back connection");
this.close();
// Let's close all connections linked to that room
for (const listener of this.listeners) {
listener.disconnecting = true;
listener.end(1011, "Connection closed between pusher and back server");
}
}
});
}
public close(): void {
debug("Closing connection to room %s on back server", this.roomUrl);
this.isClosing = true;
this.backConnection.cancel();
}
}

11
pusher/src/Model/Zone.ts
View file

@ -15,6 +15,7 @@ import {
ZoneMessage,
EmoteEventMessage,
CompanionMessage,
ErrorMessage,
} from "../Messages/generated/messages_pb";
import { ClientReadableStream } from "grpc";
import { PositionDispatcher } from "_Model/PositionDispatcher";
@ -30,6 +31,7 @@ export interface ZoneEventListener {
onGroupMoves(group: GroupDescriptor, listener: ExSocketInterface): void;
onGroupLeaves(groupId: number, listener: ExSocketInterface): void;
onEmote(emoteMessage: EmoteEventMessage, listener: ExSocketInterface): void;
onError(errorMessage: ErrorMessage, listener: ExSocketInterface): void;
}
/*export type EntersCallback = (thing: Movable, listener: User) => void;
@ -217,6 +219,9 @@ export class Zone {
} else if (message.hasEmoteeventmessage()) {
const emoteEventMessage = message.getEmoteeventmessage() as EmoteEventMessage;
this.notifyEmote(emoteEventMessage);
} else if (message.hasErrormessage()) {
const errorMessage = message.getErrormessage() as ErrorMessage;
this.notifyError(errorMessage);
} else {
throw new Error("Unexpected message");
}
@ -303,6 +308,12 @@ export class Zone {
}
}
private notifyError(errorMessage: ErrorMessage) {
for (const listener of this.listeners) {
this.socketListener.onError(errorMessage, listener);
}
}
/**
* Notify listeners of this zone that this group left
*/

22
pusher/src/Services/SocketManager.ts
View file

@ -30,6 +30,8 @@ import {
ViewportMessage,
WebRtcSignalToServerMessage,
WorldConnexionMessage,
VariableMessage,
ErrorMessage,
WorldFullMessage,
} from "../Messages/generated/messages_pb";
import { ProtobufUtils } from "../Model/Websocket/ProtobufUtils";
@ -226,6 +228,9 @@ export class SocketManager implements ZoneEventListener {
const pusherToBackMessage = new PusherToBackMessage();
pusherToBackMessage.setJoinroommessage(joinRoomMessage);
streamToPusher.write(pusherToBackMessage);
const pusherRoom = await this.getOrCreateRoom(client.roomId);
pusherRoom.join(client);
} catch (e) {
console.error('An error occurred on "join_room" event');
console.error(e);
@ -277,6 +282,13 @@ export class SocketManager implements ZoneEventListener {
emitInBatch(listener, subMessage);
}
onError(errorMessage: ErrorMessage, listener: ExSocketInterface): void {
const subMessage = new SubMessage();
subMessage.setErrormessage(errorMessage);
emitInBatch(listener, subMessage);
}
// Useless now, will be useful again if we allow editing details in game
handleSetPlayerDetails(client: ExSocketInterface, playerDetailsMessage: SetPlayerDetailsMessage) {
const pusherToBackMessage = new PusherToBackMessage();
@ -299,6 +311,13 @@ export class SocketManager implements ZoneEventListener {
client.backConnection.write(pusherToBackMessage);
}
handleVariableEvent(client: ExSocketInterface, variableMessage: VariableMessage) {
const pusherToBackMessage = new PusherToBackMessage();
pusherToBackMessage.setVariablemessage(variableMessage);
client.backConnection.write(pusherToBackMessage);
}
async handleReportMessage(client: ExSocketInterface, reportPlayerMessage: ReportPlayerMessage) {
try {
await adminApi.reportPlayer(
@ -339,6 +358,7 @@ export class SocketManager implements ZoneEventListener {
room.leave(socket);
if (room.isEmpty()) {
room.close();
this.rooms.delete(socket.roomId);
debug("Room %s is empty. Deleting.", socket.roomId);
}
@ -368,7 +388,7 @@ export class SocketManager implements ZoneEventListener {
if (ADMIN_API_URL) {
await this.updateRoomWithAdminData(room);
}
await room.init();
this.rooms.set(roomUrl, room);
}
return room;

2
pusher/tsconfig.json
View file

@ -3,7 +3,7 @@
"experimentalDecorators": true,
/* Basic Options */
// "incremental": true, /* Enable incremental compilation */
"target": "es5", /* Specify ECMAScript target version: 'ES3' (default), 'ES5', 'ES2015', 'ES2016', 'ES2017', 'ES2018', 'ES2019', 'ES2020', or 'ESNEXT'. */
"target": "ES2019", /* Specify ECMAScript target version: 'ES3' (default), 'ES5', 'ES2015', 'ES2016', 'ES2017', 'ES2018', 'ES2019', 'ES2020', or 'ESNEXT'. */
"downlevelIteration": true,
"module": "commonjs", /* Specify module code generation: 'none', 'commonjs', 'amd', 'system', 'umd', 'es2015', 'es2020', or 'ESNext'. */
// "lib": [], /* Specify library files to be included in the compilation. */