service-msg/app/service/rabbitmq.js

'use strict';
const msgvalue = require('../util/constants');
const Service = require('egg').Service;

class RabbitmqService extends Service {

  constructor(ctx) {
    super(ctx);
    this.exType = 'topic';
    this.durable = true;
  }

  // 发送消息
  async sendQueueMsg(ex, routeKey, msg, parm) {
    const { mq } = this.ctx;
    if (mq) {
      await mq.topic(ex, routeKey, msg, parm);
    } else {
      this.ctx.logger.error('!!!!!!没有配置MQ插件!!!!!!');
    }
  }

  // 接收消息
  async receiveQueueMsg(ex) {
    const self = this;
    const { mq } = self.ctx;
    if (mq) {
      const ch = await mq.conn.createChannel();
      try {
        await ch.assertExchange(ex, self.exType, { durable: self.durable });
        const q = await ch.assertQueue('', { exclusive: true });
        console.log('==q=', q);
        // 队列绑定 exchange
        await ch.bindQueue(q.queue, ex, '*');
        ch.consume(q.queue, msg => {
          console.log('收到消息: ', msg);
          const result = msg.content.toString();
          const headers = msg.properties.headers;
          // 插入待办事项到数据库中。
          if (result != null) {
            this.service.message.create({ userid: headers.userId, name: headers.name, createtime: headers.createtime, type: headers.type, content: result, remark: headers.remark });
          }
        }, { noAck: true });
      } catch (e) {
        console.log('==e==', e);
        await ch.close();
      }
    } else {
      this.ctx.logger.error('!!!!!!没有配置MQ插件!!!!!!');
    }
  }

  // 接收学校审核消息并发送模板消息到企业微信
  async receiveMsgSendWxSch(ex) {
    const self = this;
    const { mq } = self.ctx;
    if (mq) {
      const ch = await mq.conn.createChannel();
      try {
        await ch.assertExchange(ex, self.exType, { durable: self.durable });
        const q = await ch.assertQueue('', { exclusive: true });
        console.log('==q=', q);
        // 队列绑定 exchange
        await ch.bindQueue(q.queue, ex, '*');
        ch.consume(q.queue, msg => {
          console.log('收到消息: ', msg);
          const result = msg.content.toString();
          // const fields = msg.fields;
          // const properties = msg.properties;
          const headers = msg.properties.headers;
          // 插入待办事项到数据库中。
          const path = self.ctx.app.config.baseDirMq + self.ctx.app.config.corpsDirMq + headers.userId + '/users';
          self.ctx.curl(path, {
            method: 'GET',
            dataType: 'json',
          }).then(
            function(corpUser) {
              if (corpUser != null) {
                const _datas = corpUser.data.data;
                const _self = self;
                for (let i = 0; i < _datas.length; i++) {
                  const _data = _datas[i];
                  self.service.wechat.sendTemplateMsg(_self.ctx.app.config.REVIEW_TEMPLATE_ID, _data.openid, '您的申请已经得到批复', result, headers.name, '企业申请', headers.remark);
                }
              }
            });
        }, { noAck: true });
      } catch (e) {
        console.log('==e==', e);
        await ch.close();
      }
    } else {
      this.ctx.logger.error('!!!!!!没有配置MQ插件!!!!!!');
    }
  }

  // 接收企业职位发布消息并发送模板消息到学生微信
  async receiveMsgSendWxCorp(ex) {
    const self = this;
    const { mq } = self.ctx;
    if (mq) {
      const ch = await mq.conn.createChannel();
      try {
        await ch.assertExchange(ex, self.exType, { durable: self.durable });
        const q = await ch.assertQueue('', { exclusive: true });
        console.log('==q=', q);
        // 队列绑定 exchange
        await ch.bindQueue(q.queue, ex, '*');
        ch.consume(q.queue, msg => {
          console.log('收到消息: ', msg);
          const result = msg.content.toString();
          // const fields = msg.fields;
          // const properties = msg.properties;
          const headers = msg.properties.headers;
          // 插入待办事项到数据库中。
          const path = self.ctx.app.config.baseDirMq + self.ctx.app.config.stusDirMq + '?corpid=' + headers.userId;
          // const path = 'http://10.16.5.15:8101/api/studentcorp' + headers.userid;
          self.ctx.curl(path, {
            method: 'GET',
            dataType: 'json',
          }).then(
            function(stus) {
              if (stus != null) {
                for (const elem of stus.data.data) {
                  const pathStu = self.ctx.app.config.baseDirMq + self.ctx.app.config.strDirMq + elem.studid;
                  self.ctx.curl(pathStu, {
                    method: 'GET',
                    dataType: 'json',
                  }).then(
                    // eslint-disable-next-line no-loop-func
                    function(stud) {
                      if (stud != null) {
                        self.service.wechat.sendTemplateMsg(self.ctx.app.config.REVIEW_TEMPLATE_ID, stud.data.data.openid, '您订阅的企业已经有新职位发布', result, headers.name, '职位发布', headers.remark);
                      }
                    });
                }
              }
            });


        }, { noAck: true });
      } catch (e) {
        console.log('==e==', e);
        await ch.close();
      }
    } else {
      this.ctx.logger.error('!!!!!!没有配置MQ插件!!!!!!');
    }
  }

  // 接收企业发送面试邀请发送给学生消息
  async receiveQueueMsgOffer(ex) {
    const self = this;
    const { mq } = self.ctx;
    if (mq) {
      const ch = await mq.conn.createChannel();
      try {
        await ch.assertExchange(ex, self.exType, { durable: self.durable });
        const q = await ch.assertQueue('', { exclusive: true });
        console.log('==q=', q);
        // 队列绑定 exchange
        await ch.bindQueue(q.queue, ex, '*');
        ch.consume(q.queue, msg => {
          console.log('收到消息: ', msg);
          const result = msg.content.toString();
          // const fields = msg.fields;
          // const properties = msg.properties;
          const headers = msg.properties.headers;
          // 插入待办事项到数据库中。
          const pathStu = self.ctx.app.config.baseDirMq + self.ctx.app.config.strDirMq + headers.userId;
          self.ctx.curl(pathStu, {
            method: 'GET',
            dataType: 'json',
          }).then(
            // eslint-disable-next-line no-loop-func
            function(stud) {
              if (stud != null) {
                self.service.wechat.sendTemplateMsg(self.ctx.app.config.REVIEW_TEMPLATE_ID, stud.data.data.openid, '您有新的面试邀请，请及时查看。', result, headers.name, '面试邀请', headers.remark);
              }
            });
        }, { noAck: true });
      } catch (e) {
        console.log('==e==', e);
        await ch.close();
      }
    } else {
      this.ctx.logger.error('!!!!!!没有配置MQ插件!!!!!!');
    }
  }

}

module.exports = RabbitmqService;