dh_demo

event_queue.mjs (4181B)

---

 #!/usr/bin/env node
 
 // Event Queue는 Redis에서 이벤트를 구독하고, 웹소켓 클라이언트에
 // 이벤트를 전달하는 작은 프로그램입니다.
 //
 // 다음 환경 변수를 사용합니다.
 //  REDIS_URL      - Redis 서버의 URL입니다.
 //  REDIS_USERNAME - Redis 서버의 사용자 이름입니다.
 //  REDIS_PASSWORD - Redis 서버의 비밀번호입니다.
 //  REDIS_DB       - Redis 서버의 데이터베이스 번호입니다.
 //  WSS_PORT       - 웹소켓 서버의 포트 번호입니다.
 //
 // 다음 채널을 구독합니다.
 //  event
 // 이벤트 페이로드의 형식은 다음과 같습니다.
 //  [path]|[event]|[payload]
 
 import redis from 'redis'
 import { WebSocketServer } from 'ws'
 import { nanoid } from 'nanoid'
 
 /**
  * @typedef {string} Path
  * @typedef {string} SocketID
  */
 
 /**
  * 웹소켓 접속을 관리합니다.
  */
 class SocketManager {
   constructor () {
     this.server = new WebSocketServer({
       port: parseIntOrDefault(process.env.WSS_PORT, 3001),
       clientTracking: false,
     })
 
     /**
      * @type {Map<Path, Set<SocketID>>}
      */
     this.paths = new Map()
 
     /**
      * @type {Map<SocketID, Path>}
      */
     this.socketPaths = new Map()
 
     /**
      * @type {Map<SocketID, WebSocket>}
      */
     this.sockets = new Map()
 
     // Setup socket handlers
     this.server.on('connection', ws => {
       this.#handleConnection(ws)
     })
   }
 
   /**
    * @param ws {WebSocket}
    */
   #handleConnection (ws) {
     const id = nanoid()
     this.sockets.set(id, ws)
 
     console.log(`Connected: ${id}`)
 
     ws.on('message', (message, isBinary) => {
       this.#handleMessage(id, ws, message, isBinary)
     })
     ws.on('close', () => {
       this.#unregister(id)
     })
 
     setTimeout(() => {
       ws.send('HAND')
     }, 500)
   }
 
   /**
    * @param id {SocketID}
    */
   #unregister (id) {
     const path = this.socketPaths.get(id)
     if (this.paths.has(path)) {
       this.paths.get(path).delete(id)
     }
     this.socketPaths.delete(id)
   }
 
   /**
    * @param message {import('ws').WebSocket.RawData}
    * @return {[string, string]}
    */
   #parseCommand (message) {
     const src = message.toString().trim()
     return [src.slice(0, 4), src.slice(4)] // Command is always 4 characters
   }
 
   /**
    * @param id {SocketID}
    * @param ws {import('ws').WebSocket}
    * @param message {import('ws').WebSocket.RawData}
    * @param isBinary {boolean}
    */
   #handleMessage (id, ws, message, isBinary) {
     if (!this.sockets.has(id)) {
       return
     }
 
     const [command, payload] = this.#parseCommand(message)
     switch (command) {
       case 'PATH':
         this.#changePath(id, payload)
         ws.send('OKAY')
         break
 
       case 'TEST':
         this.broadcast(payload, 'TEST')
         break
 
       default: {
         ws.send('????')
         break
       }
     }
   }
 
   /**
    * @param id {SocketID}
    * @param path {Path}
    */
   #changePath (id, path) {
     const oldPath = this.socketPaths.get(id)
     if (oldPath) {
       this.paths.get(oldPath).delete(id)
     }
 
     if (!this.paths.has(path)) {
       this.paths.set(path, new Set())
     }
     this.paths.get(path).add(id)
   }
 
   /**
    * @param path {Path}
    * @param message {string}
    */
   broadcast (path, message) {
     if (!this.paths.has(path)) {
       return
     }
     this.paths.get(path).forEach(id => {
       const ws = this.sockets.get(id)
       ws.send(message)
     })
   }
 }
 
 function parseIntOrDefault (value, defaultValue) {
   const parsed = parseInt(value, 10)
   return Number.isNaN(parsed) ? defaultValue : parsed
 }
 
 !(async () => {
   const redisClient = redis.createClient({
     url: process.env.REDIS_URL,
     username: process.env.REDIS_USERNAME,
     password: process.env.REDIS_PASSWORD,
     database: parseIntOrDefault(process.env.REDIS_DB, undefined),
   })
 
   const socketManager = new SocketManager()
 
   await redisClient.connect()
 
   await redisClient.subscribe('event', (message) => {
     const [path, event, payload] = message.split('|')
     socketManager.broadcast(path, `EMIT${event} ${payload}`)
   })
 
   console.log('Started')
 })()
   .catch(err => {
     console.error(err)
     process.exit(1)
   })