dros/src/domain/mqttServiceForDevice.ts

import mqtt from 'mqtt';
import { MQTT_BROKER_URL } from '../API/config';
import emitter from '../utils/eventEmitter';

interface MqttMessage {
  IDX: string;
  TYPE: string;
  CMD: string;
  KEY: string;
  CONTEXT: string;
  Ver: string;
  Transaction: string;
  HANDLE: {
    ROUTE: string;
    FLAGS: string;
    LANG: string;
    HANDLEID: string;
    OWNERID: {
      EBUSID: string;
      MACHINEID: string;
      PROCID: string;
      ADDR: string;
    };
    DEVID: {
      EBUSID: string;
      MACHINEID: string;
      PROCID: string;
      ADDR: string;
    };
  };
  TOPIC: string;
  PUBLISH: string;
}

const MQTT_TOPIC_2 = 'CCOS/DEVICE/Generator/Notify/GENERATORSYNCSTATE'; //发生器
const MQTT_TOPIC = 'CCOS/DEVICE/Detector/Notify/XrayON';
const MQTT_TOPIC_DETECTOR = 'CCOS/DEVICE/Detector/Notify/DetectorStatus'; //探测器

const options = {
  clean: true,
  connectTimeout: 4000,
  username: 'mqttx1112',
  password: '',
};

let mqttClient;

const handleMqttMessage = (message: MqttMessage) => {
  console.log(`[mqttServiceForDevice] 收到message.CONTEXT ${message.CONTEXT}`);
  if (message.CONTEXT === '0') {
    emitter.emit('GENERATOR_RAD_OFF');
  } else if (message.CONTEXT === '1') {
    emitter.emit('ACQUISITION_STARTED');
    emitter.emit('GENERATOR_RAD_PREPARE');
  } else if (message.CONTEXT === '2') {
    emitter.emit('GENERATOR_RAD_READY');
  } else if (message.CONTEXT === '3') {
    emitter.emit('ACQUISITION_STARTED');
  }
};
const handleMqttMessageFromDetector = (message: MqttMessage) => {
  console.log(
    `[mqttServiceForDevice] 收到来自于Detector 的 message.CONTEXT ${message.CONTEXT}`
  );
  if (message.CONTEXT === '4') {
    //探测器状态会被通知为4（表示准备就绪）
    emitter.emit('DETECTOR_ACQUISITION_STARTED');
  } else if (message.CONTEXT === '5') {
    //探测器状态首先被通知为5（表示正在采集）
    emitter.emit('DETECTOR_ACQUISITION_INPROGRESS');
  } else {
    //探测器出错
    emitter.emit('DETECTOR_ACQUISITION_ERROR');
  }
};

const startListening = () => {
  mqttClient = mqtt.connect(MQTT_BROKER_URL, options);
  mqttClient.on('connect', () => {
    console.log('[mqttServiceForDevice] Connected to MQTT broker');
    mqttClient.subscribe(MQTT_TOPIC, (err) => {
      if (err) {
        console.error(
          '[mqttServiceForDevice] Failed to subscribe to topic',
          err
        );
      }
    });
    mqttClient.subscribe(MQTT_TOPIC_2, (err) => {
      if (err) {
        console.error(
          '[mqttServiceForDevice] Failed to subscribe to topic',
          err
        );
      }
    });
    mqttClient.subscribe(MQTT_TOPIC_DETECTOR, (err) => {
      if (err) {
        console.error(
          '[mqttServiceForDevice] Failed to subscribe to topic MQTT_TOPIC_DETECTOR',
          err
        );
      }
    });
  });

  mqttClient.on('error', (error) => {
    console.log('[mqttServiceForDevice] Connection failed:', error);
  });

  mqttClient.on('message', (topic, message) => {
    if (topic === MQTT_TOPIC) {
      const parsedMessage: MqttMessage = JSON.parse(message.toString());
      handleMqttMessage(parsedMessage);
    }
    if (topic === MQTT_TOPIC_2) {
      const parsedMessage: MqttMessage = JSON.parse(message.toString());
      handleMqttMessage(parsedMessage);
    }
    if (topic === MQTT_TOPIC_DETECTOR) {
      const parsedMessage: MqttMessage = JSON.parse(message.toString());
      handleMqttMessageFromDetector(parsedMessage);
    }
  });
};

const stopListening = () => {
  mqttClient.end();
  console.log('[mqttServiceForDevice] Stopped listening to MQTT broker');
};

export { startListening, stopListening };