Mass replace parameter order (socket,io)>(io,socket), worked on queueing system

This commit is contained in:
Dennis Postma 2024-09-21 23:54:52 +02:00
parent 10dc9df8a9
commit 9d6de8a1a9
36 changed files with 206 additions and 121 deletions

View File

@ -0,0 +1,16 @@
import { TSocket } from '../utilities/types'
import { Server as SocketServer } from 'socket.io'
export default class SomeJob {
constructor(private params: any) {}
async execute(io: SocketServer, socket?: TSocket) {
// Handle the event
console.log('Event params:', this.params)
if (socket) {
socket.emit('notification', { message: 'Something happened with socket' })
}
// Use io for broadcasting if needed
io.emit('notification', { message: 'Something happened' })
}
}

View File

@ -31,7 +31,7 @@ class CharacterManager {
}
public hasResetMovement(character: ExtendedCharacter) {
return this.characters.find(x => x.id === character.id)?.resetMovement;
return this.characters.find((x) => x.id === character.id)?.resetMovement
}
public getCharactersInZone(zone: Zone) {

View File

@ -1,29 +1,91 @@
import IORedis from 'ioredis';
import IORedis from 'ioredis'
import { Job, Queue, Worker } from 'bullmq'
import config from '../utilities/config'
import { Server as SocketServer } from 'socket.io'
import { TSocket } from '../utilities/types'
import { queueLogger } from '../utilities/logger'
class QueueManager {
private connection!: IORedis;
public queue!: Queue;
public worker!: Worker;
private connection!: IORedis
private queue!: Queue
private worker!: Worker
private io!: SocketServer
public async boot(io: SocketServer) {
this.io = io
public async boot() {
this.connection = new IORedis(config.REDIS_URL, {
maxRetriesPerRequest: null
});
// this.queue = new Queue('myqueue', { connection: this.connection });
// this.worker = new Worker('myqueue', async (job: Job)=>{
// console.log('hallo')
// console.log(job.data);
// console.log(job.data.data.classobj);
// const test = job.data.data.classobj();
// console.log(test);
// console.log(job.data);
// }, { connection: this.connection, concurrency: 10000 });
//
})
console.log('hallo')
try {
await this.connection.ping()
queueLogger.info('Successfully connected to Redis')
} catch (error) {
queueLogger.error('Failed to connect to Redis:', error)
process.exit(1)
}
this.queue = new Queue('jobs')
this.worker = new Worker('jobs', this.processJob.bind(this), {
connection: this.connection,
concurrency: 10000
})
this.worker.on('completed', (job) => {
queueLogger.info(`Job ${job?.id} has completed`)
})
this.worker.on('failed', (job, err) => {
queueLogger.error(`Job ${job?.id} failed with error: ${err}`)
})
queueLogger.info('Queue manager loaded')
}
private async processJob(job: Job) {
const { jobClass, params, socketId } = job.data
console.log('Processing job:', job)
try {
const JobClass = await import(`../jobs/${jobClass}`)
if (!JobClass) {
queueLogger.warn(`Job class not found: ${jobClass}`)
return
}
console.log('Job class:', JobClass)
console.log('Job class:', JobClass.name)
const job = new JobClass(params)
if (socketId && this.io) {
const socket = this.io.sockets.sockets.get(socketId)
if (socket) {
await job.execute(this.io, socket)
} else {
queueLogger.warn(`Socket not found for job: ${socketId}`)
await job.execute(this.io)
}
} else {
await job.execute(this.io)
}
} catch (error: any) {
queueLogger.error(`Error processing job: ${error.message}`)
}
}
export default new QueueManager();
public async addToQueue(jobClass: any, params: any, socket?: TSocket) {
const jobData = {
jobClass: jobClass.name,
params,
socketId: socket?.id
}
await this.queue.add('job', jobData)
}
}
export default new QueueManager()

View File

@ -1,6 +1,6 @@
import { Zone, ZoneEventTile, ZoneEventTileType, ZoneObject } from '@prisma/client'
import prisma from '../utilities/prisma'
import { ZoneEventTileWithTeleport } from '../events/zone/characterMove'
import { ZoneEventTileWithTeleport } from '../socketEvents/zone/characterMove'
import { appLogger } from '../utilities/logger'
class ZoneRepository {

View File

@ -58,14 +58,15 @@ export class Server {
appLogger.error(`Socket.IO failed to start: ${error.message}`)
}
// Load queue manager
await QueueManager.boot(this.io)
// Add http API routes
await addHttpRoutes(this.app)
// Load user manager
await UserManager.boot()
await QueueManager.boot();
// Load zoneEditor manager
await ZoneManager.boot()
@ -85,7 +86,7 @@ export class Server {
* @private
*/
private async handleConnection(socket: TSocket) {
const eventsPath = path.join(__dirname, 'events')
const eventsPath = path.join(__dirname, 'socketEvents')
try {
await this.loadEventHandlers(eventsPath, socket)
} catch (error: any) {
@ -112,7 +113,7 @@ export class Server {
eventInstance.listen()
} else {
// This is a function-based event
module.default(socket, this.io)
module.default(this.io, socket)
}
} else {
appLogger.warn(`Unrecognized export in ${file.name}`)

View File

@ -2,19 +2,22 @@ import { Server } from 'socket.io'
import { TSocket, ExtendedCharacter } from '../../utilities/types'
import CharacterRepository from '../../repositories/characterRepository'
import CharacterManager from '../../managers/characterManager'
import QueueManager from '../../managers/queueManager'
import SomeJob from '../../jobs/characterLeaveZone'
type SocketResponseT = {
character_id: number
}
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('character:connect', async (data: SocketResponseT) => {
console.log('character:connect requested', data)
try {
const character = await CharacterRepository.getByUserAndId(socket?.user?.id as number, data.character_id);
if(!character) return;
socket.characterId = character.id;
CharacterManager.initCharacter(character as ExtendedCharacter);
const character = await CharacterRepository.getByUserAndId(socket?.user?.id as number, data.character_id)
if (!character) return
socket.characterId = character.id
await QueueManager.addToQueue(SomeJob, { someParam: 'value' }, socket)
CharacterManager.initCharacter(character as ExtendedCharacter)
socket.emit('character:connect', character)
} catch (error: any) {
console.log('character:connect error', error)

View File

@ -6,7 +6,7 @@ import { ZCharacterCreate } from '../../utilities/zodTypes'
import prisma from '../../utilities/prisma'
import { gameLogger } from '../../utilities/logger'
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('character:create', async (data: any) => {
console.log('character:create requested', data)
// zod validate

View File

@ -12,7 +12,7 @@ type TypeResponse = {
characters: Character[]
}
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('character:delete', async (data: TypePayload, callback: (response: TypeResponse) => void) => {
// zod validate
try {

View File

@ -3,7 +3,7 @@ import { TSocket } from '../../utilities/types'
import { Character } from '@prisma/client'
import CharacterRepository from '../../repositories/characterRepository'
export default function CharacterList(socket: TSocket, io: Server) {
export default function CharacterList(io: Server, socket: TSocket) {
socket.on('character:list', async (data: any) => {
try {
console.log('character:list requested')

View File

@ -7,7 +7,7 @@ type TypePayload = {
message: string
}
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('chat:send_message', async (data: TypePayload, callback: (response: boolean) => void) => {
try {
if (!isCommand(data.message, 'alert')) return

View File

@ -21,7 +21,7 @@ export default class TeleportCommandEvent {
private async handleTeleportCommand(data: TypePayload, callback: (response: boolean) => void): Promise<void> {
try {
const character = CharacterManager.getCharacterFromSocket(this.socket);
const character = CharacterManager.getCharacterFromSocket(this.socket)
if (!character) {
this.socket.emit('notification', { title: 'Server message', message: 'Character not found' })
return
@ -61,12 +61,11 @@ export default class TeleportCommandEvent {
this.io.to(zone.id.toString()).emit('zone:character:join', character)
this.socket.join(zone.id.toString())
character.zoneId = zone.id
character.positionX = 0
character.positionY = 0
character.resetMovement = true;
character.resetMovement = true
this.socket.emit('zone:character:teleport', {
zone,

View File

@ -2,7 +2,7 @@ import { Server } from 'socket.io'
import { TSocket } from '../utilities/types'
import CharacterManager from '../managers/characterManager'
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('disconnect', async (data: any) => {
if (!socket.user) {
console.log('User disconnected but had no user set')
@ -11,7 +11,7 @@ export default function (socket: TSocket, io: Server) {
io.emit('user:disconnect', socket.user.id)
const character = CharacterManager.getCharacterFromSocket(socket);
const character = CharacterManager.getCharacterFromSocket(socket)
if (!character) {
console.log('User disconnected but had no character set')

View File

@ -12,9 +12,9 @@ interface IPayload {}
* @param socket
* @param io
*/
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('gm:object:list', async (data: any, callback: (response: Object[]) => void) => {
const character = await characterRepository.getById(socket.characterId as number);
const character = await characterRepository.getById(socket.characterId as number)
if (!character) return callback([])
if (character.role !== 'gm') {

View File

@ -15,9 +15,9 @@ interface IPayload {
* @param socket
* @param io
*/
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('gm:object:remove', async (data: IPayload, callback: (response: boolean) => void) => {
const character = await characterRepository.getById(socket.characterId as number);
const character = await characterRepository.getById(socket.characterId as number)
if (!character) return callback(false)
if (character.role !== 'gm') {

View File

@ -21,9 +21,9 @@ type Payload = {
* @param socket
* @param io
*/
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('gm:object:update', async (data: Payload, callback: (success: boolean) => void) => {
const character = await characterRepository.getById(socket.characterId as number);
const character = await characterRepository.getById(socket.characterId as number)
if (!character) return callback(false)
if (character.role !== 'gm') {

View File

@ -24,7 +24,7 @@ export default class ObjectUploadEvent {
private async handleObjectUpload(data: IObjectData, callback: (response: boolean) => void): Promise<void> {
try {
const character = await characterRepository.getById(this.socket.characterId as number);
const character = await characterRepository.getById(this.socket.characterId as number)
if (!character) return callback(false)
if (character.role !== 'gm') {

View File

@ -11,10 +11,10 @@ import characterRepository from '../../../../repositories/characterRepository'
* @param socket
* @param io
*/
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('gm:sprite:create', async (data: undefined, callback: (response: boolean) => void) => {
try {
const character = await characterRepository.getById(socket.characterId as number);
const character = await characterRepository.getById(socket.characterId as number)
if (!character) return callback(false)
if (character.role !== 'gm') {

View File

@ -25,7 +25,7 @@ export default class GMSpriteDeleteEvent {
}
private async handleSpriteDelete(data: Payload, callback: (response: boolean) => void): Promise<void> {
const character = CharacterManager.getCharacterFromSocket(this.socket);
const character = CharacterManager.getCharacterFromSocket(this.socket)
if (character?.role !== 'gm') {
return callback(false)
}

View File

@ -12,9 +12,9 @@ interface IPayload {}
* @param socket
* @param io
*/
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('gm:sprite:list', async (data: any, callback: (response: Sprite[]) => void) => {
const character = await characterRepository.getById(socket.characterId as number);
const character = await characterRepository.getById(socket.characterId as number)
if (!character) return callback([])
if (character.role !== 'gm') {

View File

@ -27,9 +27,9 @@ interface ProcessedSpriteAction extends SpriteActionInput {
}>
}
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('gm:sprite:update', async (data: Payload, callback: (success: boolean) => void) => {
const character = CharacterManager.getCharacterFromSocket(socket);
const character = CharacterManager.getCharacterFromSocket(socket)
if (character?.role !== 'gm') {
return callback(false)
}

View File

@ -25,7 +25,7 @@ export default class GMTileDeleteEvent {
}
private async handleTileDelete(data: Payload, callback: (response: boolean) => void): Promise<void> {
const character = await characterRepository.getById(this.socket.characterId as number);
const character = await characterRepository.getById(this.socket.characterId as number)
if (!character) return callback(false)
if (character.role !== 'gm') {

View File

@ -12,9 +12,9 @@ interface IPayload {}
* @param socket
* @param io
*/
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('gm:tile:list', async (data: any, callback: (response: Tile[]) => void) => {
const character = await characterRepository.getById(socket.characterId as number);
const character = await characterRepository.getById(socket.characterId as number)
if (!character) return
if (character.role !== 'gm') {

View File

@ -15,9 +15,9 @@ type Payload = {
* @param socket
* @param io
*/
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('gm:tile:update', async (data: Payload, callback: (success: boolean) => void) => {
const character = await characterRepository.getById(socket.characterId as number);
const character = await characterRepository.getById(socket.characterId as number)
if (!character) return callback(false)
if (character.role !== 'gm') {

View File

@ -16,10 +16,10 @@ interface ITileData {
* @param socket
* @param io
*/
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('gm:tile:upload', async (data: ITileData, callback: (response: boolean) => void) => {
try {
const character = await characterRepository.getById(socket.characterId as number);
const character = await characterRepository.getById(socket.characterId as number)
if (!character) return callback(false)
if (character.role !== 'gm') {

View File

@ -17,9 +17,9 @@ type Payload = {
* @param socket
* @param io
*/
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('gm:zone_editor:zone:create', async (data: Payload, callback: (response: Zone[]) => void) => {
const character = await characterRepository.getById(socket.characterId as number);
const character = await characterRepository.getById(socket.characterId as number)
if (!character) return
if (character.role !== 'gm') {

View File

@ -14,9 +14,9 @@ type Payload = {
* @param socket
* @param io
*/
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('gm:zone_editor:zone:delete', async (data: Payload, callback: (response: boolean) => void) => {
const character = await characterRepository.getById(socket.characterId as number);
const character = await characterRepository.getById(socket.characterId as number)
if (!character) return
if (character.role !== 'gm') {

View File

@ -12,9 +12,9 @@ interface IPayload {}
* @param socket
* @param io
*/
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('gm:zone_editor:zone:list', async (data: IPayload, callback: (response: Zone[]) => void) => {
const character = await characterRepository.getById(socket.characterId as number);
const character = await characterRepository.getById(socket.characterId as number)
if (!character) return
if (character.role !== 'gm') {

View File

@ -14,9 +14,9 @@ interface IPayload {
* @param socket
* @param io
*/
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('gm:zone_editor:zone:request', async (data: IPayload, callback: (response: Zone) => void) => {
const character = await characterRepository.getById(socket.characterId as number);
const character = await characterRepository.getById(socket.characterId as number)
if (!character) return
if (character.role !== 'gm') {

View File

@ -32,9 +32,9 @@ interface IPayload {
* @param socket
* @param io
*/
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('gm:zone_editor:zone:update', async (data: IPayload, callback: (response: Zone) => void) => {
const character = await characterRepository.getById(socket.characterId as number);
const character = await characterRepository.getById(socket.characterId as number)
if (!character) return
if (character.role !== 'gm') {

View File

@ -1,7 +1,7 @@
import { Server } from 'socket.io'
import { TSocket } from '../utilities/types'
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('login', () => {
if (!socket.user) return
socket.emit('logged_in', { user: socket.user })

View File

@ -19,11 +19,11 @@ interface IResponse {
* @param socket
* @param io
*/
export default function (socket: TSocket, io: Server) {
export default function (io: Server, socket: TSocket) {
socket.on('zone:character:join', async (callback: (response: IResponse) => void) => {
try {
if (!socket.characterId) return
const character = CharacterManager.getCharacterFromSocket(socket);
const character = CharacterManager.getCharacterFromSocket(socket)
if (!character) return

View File

@ -16,7 +16,7 @@ export default class ZoneLeaveEvent {
private async handleZoneLeave(): Promise<void> {
try {
const character = CharacterManager.getCharacterFromSocket(this.socket);
const character = CharacterManager.getCharacterFromSocket(this.socket)
if (!character) {
gameLogger.error('zone:character:leave error', 'Character not found')
return

View File

@ -58,7 +58,7 @@ export default class CharacterMove {
}
if (!character.isMoving && !character.resetMovement) {
character.isMoving = true
this.currentZoneId[character.id] = character.zoneId;
this.currentZoneId[character.id] = character.zoneId
await this.moveAlongPath(character, path)
}
}
@ -115,8 +115,8 @@ export default class CharacterMove {
if (this.currentZoneId[character.id] === character.zoneId) {
await this.moveAlongPath(character, this.nextPath[character.id])
} else {
delete this.currentZoneId[character.id];
character.isMoving = false;
delete this.currentZoneId[character.id]
character.isMoving = false
}
} else {
this.finalizeMovement(character)

View File

@ -84,20 +84,24 @@ async function addHttpRoutes(app: Application) {
return res.status(404).json({ message: 'Zone not found' })
}
const assets = await zoneManager.getZoneAssets(zone);
const assets = await zoneManager.getZoneAssets(zone)
res.json([
...assets.tiles.map(x => { return {
...assets.tiles.map((x) => {
return {
key: x,
url: '/assets/tiles/' + x + '.png',
group: 'tiles'
}}),
...assets.objects.map(x => { return {
}
}),
...assets.objects.map((x) => {
return {
key: x,
url: '/assets/objects/' + x + '.png',
group: 'objects'
}})
]);
}
})
])
})
/**

View File

@ -3,10 +3,11 @@ import fs from 'fs'
import path from 'path'
// Array of log types
const LOG_TYPES = ['http', 'game', 'gameMaster', 'app'] as const
type LogType = typeof LOG_TYPES[number]
const LOG_TYPES = ['http', 'game', 'gameMaster', 'app', 'queue'] as const
type LogType = (typeof LOG_TYPES)[number]
const createLogger = (name: LogType) => pino({
const createLogger = (name: LogType) =>
pino({
level: process.env.LOG_LEVEL || 'debug',
transport: {
target: 'pino/file',
@ -25,12 +26,10 @@ const createLogger = (name: LogType) => pino({
})
// Create logger instances
const loggers = Object.fromEntries(
LOG_TYPES.map(type => [type, createLogger(type)])
) as Record<LogType, ReturnType<typeof createLogger>>
const loggers = Object.fromEntries(LOG_TYPES.map((type) => [type, createLogger(type)])) as Record<LogType, ReturnType<typeof createLogger>>
const watchLogs = () => {
LOG_TYPES.forEach(type => {
LOG_TYPES.forEach((type) => {
const logFile = path.join(__dirname, '../../logs', `${type}.log`)
fs.watchFile(logFile, (curr, prev) => {
@ -44,5 +43,6 @@ const watchLogs = () => {
})
}
export const { http: httpLogger, game: gameLogger, gameMaster: gameMasterLogger, app: appLogger } = loggers
export const { http: httpLogger, game: gameLogger, gameMaster: gameMasterLogger, app: appLogger, queue: queueLogger } = loggers
export { watchLogs }