[TIL] 24.11.19 TUE

큐를 이용한 패킷 처리

기존 패킷 전송은 각 핸들러에서 socket.write 하는 방식이었는데 이번에는 추후 상태관리 및 확장성을 염두하여
싱글턴 패턴의 클래스로 관리 해보고자 했다.
완성되고 테스트 해봐야 아는 부분이겠지만 그 전 대규모 처리에 대한 자료를 조금 더 찾아봐서 리팩토링 할 예정이다.

큐 방식의 장점

우선 큐를 도입하면 순차적 처리가 보장되고 전송할 패킷이 한번에 많이 몰려도 순서대로 처리할 수 있다.
큐에 패킷을 쌓아놓고 비동기적으로 처리하면 주요 로직이 블로킹되지 않을 것이다.

단점

처리할 패킷이 많아지면 시간이 지연되고 관리 로직이 추가되어 복잡성이 증가되며 우선순위가 높은 경우를 대비해 따로 큐를 도입해야하는 단점이 있다.

소규모 프로젝트이기 때문에 소켓 처리 방식을 중간에 바꿔도 크게 문제가 될 것이 없다고 보기 때문에 시도 해 보고자 한다.

싱클턴 패턴을 적용한 sendPacket

아직 받는 부분은 구현하진 않았지만 우선 바로 사용할 수 있게 보내는 것부터 구현했다.

import logger from '../../utils/logger.js';
import SendPacket from './sendPacket.class';

class SendPacket {
  constructor() {
    if (SendPacket.instance instanceof SendPacket) return SendPacket.instance;
    SendPacket.instance = this;
    this.queue = [];
    this.isSending = false;
  }

  enQueue(socket, packet) {
    this.queue.push({ socket, packet });
    this.processQueue();
  }

  async processQueue() {
    if (this.isSending || !this.queue.length) return;
    this.isSending = true;

    while (this.queue.length) {
      const { socket, packet } = this.queue.shift();
      try {
        await this.sendPacket(socket, packet);
      } catch (err) {
        logger.error('Error sending packet:', err);
      }
    }

    this.isSending = false;
  }

  sendPacket(socket, packet) {
    return new Promise((resolve, reject) => {
      socket.write(packet, (err) => {
        if (err) {
          reject(err);
        } else {
          resolve();
        }
      });
    });
  }
}
const sendPacket = new SendPacket();
Object.freeze(sendPacket);

export default sendPacket;

sendPacket.enQueue(socket,packet) 으로 처리할 소켓과 패킷을 입력받으면 큐에 푸쉬하고 프로세싱 하는 방식이다.
sendPacket은 Promise를 적용하여 비동기적 처리를 진행한다.
여기서 실패시 재전송 로직이나 에러처리 부분을 보강해야한다.

위 방식을 도입한 CheckPoint 클래스

어제 TODO로 남겨놓았던 패킷 전송부분을 위 방식을 통해 구현해보았다

class CheckPoint {
  #users;
  #status;
  currentStatus;

  constructor(playerA, playerB, isTop) {
    this.isTop = isTop; // top 여부
    this.name = this.isTop ? 'top' : 'bottom'; // 콘솔 임시용
    this.playerA = playerA; // 플레이어 A 객체
    this.playerB = playerB; // 플레이어 B 객체
    this.#users = [[], []]; // 0: 팀 0, 1: 팀 1
    this.#status = 'waiting';
    this.currentStatus = this.#status; // 점령 초기화 시 상태 원복을 위한 변수
    this.timer = new Timer(5000, this.completeOccupation.bind(this)); // 5초 타이머, 콜백함수: 점령완료
  }

  // 유닛 추가/제거
  modifyUnit(team, unit, action) {
    const array = this.#users[team];

    switch (action) {
      case 'add':
        array.push(unit);
        this.checkStatus(team); // 상태 확인 (waiting, occupied(0,1), attempting(0,1))
        break;
      case 'remove':
        array.splice(array.indexOf(unit), 1);
        this.checkStatus(team);
        break;
      default:
        break;
    }
  }

  checkStatus(team) {
    const myUnits = this.getUsersCount(team); // 내 유닛 수
    const opponentUnits = this.getUsersCount(1 - team); // 상대 유닛 수
    const timerStatus = this.timer.status; // 타이머 상태

    /*
    상태에 따른 행동
    waiting - 내 유닛이 있으면 점령 시도 (동시에 적팀도 들어갈 확률은 없음, 패킷도착순)
    occupied(0,1) - 점령 상태 기준에서 상대방 유닛만 있을 경우 점령시도
    attempting(0,1) - 점령 시도 중 상대방 유닛이 들어올 경우 일시정지

    handleAttempt 함수에서 상태에 따른 행동 처리
    */
    const actions = {
      waiting: () => {
        if (myUnits) this.attemptedOccupation(team);
      },
      [`occupied${team}`]: () => {
        if (!myUnits && opponentUnits) this.attemptedOccupation(1 - team);
      },
      [`occupied${1 - team}`]: () => {
        if (myUnits && !opponentUnits) this.attemptedOccupation(team);
      },
      [`attempting${team}`]: () => {
        this.handleAttempt(opponentUnits, myUnits, timerStatus);
      },
      [`attempting${1 - team}`]: () => {
        this.handleAttempt(myUnits, opponentUnits, timerStatus);
      },
    };

    const action = actions[this.#status];
    if (action) action();
  }

  handleAttempt(opponentUnits, myUnits, timerStatus) {
    if (opponentUnits && timerStatus) {
      this.pauseOccupation();
    } else if (!opponentUnits && !timerStatus) {
      this.resumeOccupation();
    } else if (!myUnits) {
      this.clearOccupation();
      this.#status = this.currentStatus;
    }
  }

  getUsersCount(team) {
    return this.#users[team].length;
  }

  attemptedOccupation(team) {
    this.#status = `attempting${team}`;
    console.log(`${team} 팀 ${this.name}점령시도중 현재 상태: ${this.#status}`);
    this.timer.start();

    // 점령 시도 패킷 전송
    this.sendOccupationPacket(PACKET_TYPE.TRY_OCCUPATION_NOTIFICATION, false, team);
  }

  clearOccupation() {
    this.#status = this.currentStatus;
    console.log(`${this.name}점령초기화 현재 상태: ${this.#status}`);
    this.timer.allClear();

    // 점령 타이머 초기화 패킷 전송
    this.sendOccupationPacket(PACKET_TYPE.OCCUPATION_TIMER_RESET_NOTIFICATION, true);
  }

  pauseOccupation() {
    console.log(`${this.name}점령일시정지 현재 상태: ${this.#status}`);
    this.timer.pause();

    // 점령 일시정지 패킷 전송
    this.sendOccupationPacket(PACKET_TYPE.PAUSE_OCCUPATION_NOTIFICATION, true);
  }

  resumeOccupation() {
    console.log(`${this.name}점령재개 현재 상태: ${this.#status}`);
    this.timer.resume();

    // 점령 재개 패킷 전송
    const attemptingTeam = Number(this.#status.replace('attempting', '')); // attempting0 -> 0
    this.sendOccupationPacket(PACKET_TYPE.TRY_OCCUPATION_NOTIFICATION, false, attemptingTeam);
  }

  completeOccupation() {
    const completeTeam = Number(this.#status.replace('attempting', ''));
    this.#status = `occupied${this.#status.replace('attempting', '')}`;
    console.log(`${this.name}점령완료 현재 상태: ${this.#status}`);
    this.currentStatus = this.#status;
    this.timer.allClear();

    //점령완료 패킷 전송
    this.sendOccupationPacket(PACKET_TYPE.OCCUPATION_SUCCESS_NOTIFICATION, false, completeTeam);
  }

  // 중복 코드로 인한 메서드화 (책임 구분을 위해 밖으로 뺄 지 고민중)
  sendOccupationPacket(packetType, payload, target = false) {
    // payload false 인 상황이면 payloadB를 사용한다는 뜻으로 target 지정이 필요
    if (!payload && !target) throw new Error('payload or target is required');
    for (let i = 0; i < 2; i++) {
      const payloadA = { isTop: this.name === 'top' };
      const payloadB = { isTop: this.name === 'top', isOpponent: i === target ? false : true };
      const packet = createResponse(
        packetType,
        this[`player${i}`].socket.sequence++,
        payload ? payloadA : payloadB,
      );
      sendPacket.enQueue(this[`player${i}`].socket, packet);
    }
  }
}