🚀 Socket Launch Week Day 5:Introducing Repository Access Permissions and Custom Roles.Learn more →

c4-mq

Advanced tools

Install Socket

Detect and block malicious and high-risk dependencies

Install

c4-mq

C4FrameworkMQ

Source

npm

Version: 0.0.2

Version published: 8 years ago

Weekly downloads: 16

Maintainers: 1

Weekly downloads

Created: 5 years ago

Source

C4MQ

模块介绍

C4MQ是基于amqp模块构建的MQ操作类（模块），并提供了更高级的封装形式：

提供Publisher和Subscriber定义，建立在Exchange和Queue的基础上；
提供注解方式定义消息处理方法；
支持Subscriber与Publisher之间的自由绑定；
Subscriber支持对预读取的消息进行排队处理，确保处理方法的FIFO执行

使用

使用MQHandler注解对方法进行标记，提供消息处理方法；

注解项包含：

  {
      PublisherName ?: string,        // Exchange或Publisher的名字
      RoutingKey ?: string,           // routingkey
      MsgType : string                // Message的类型
  }

使用MsgBody注解对消息体中属性进行标记，用于参数绑定；
- 注解项包含：
```
{
  value : string          // 需要绑定的属性名
}
```
具体示例参考./src/MQHandlers/Hello.ts

类s

C4MQ

说明：用于连接MQ的类，对应RabbitMQ中Channel
路径：./src/C4MQ.ts
成员变量：
- m_Channel，AMQPClient
- m_Logger，日志对象(可使用console，推荐使用c4logger)

成员方法：

init

/**
 * 初始化，用于配置Channel和连接MQ
 * @param config ConnectionOptions
 * @param logger 日志对象
 */
async init(config : ConnectionOptions, logger ?: any)

disconnect

/**
 * 断开连接
 */
async disconnect()

getChannel

/**
 * 获取mq的Channel对象
 */
getChannel()

isInit

/**
 * 是否初始化
 */
isInit()

C4Exchange

说明：对RabbitMQ的Exchange的封装
路径：./src/C4Exchange.ts
成员变量：
- m_Channel，AMQPClient，连接
- m_Exchange，AMQPExchange
- m_Name，名字
- m_Logger，日志对象

成员方法：

declared

/**
 * 声明Exchange
 * @param Conn C4MQ，MQ连接
 * @param name Exchange名
 * @param options ExchangeOptions，配置属性
 * @param logger 日志对象
 */
async declared(Conn : C4MQ, name : string, options : ExchangeOptions, logger ?: any)

publish

/**
 * 发布消息
 * @param routingKey routingKey
 * @param message message
 * @param options ExchangePublishOptions，配置项
 */
async publish(routingKey : string, message : Buffer | {}, options : ExchangePublishOptions)

destroy

/**
 * 销毁
 * @param ifUnused 是否在不再使用时销毁
 */
destroy(ifUnused = true)

bindExchange

/**
 * 绑定到另一个Exchange
 * @param srcExchangeName exchange的名字
 * @param routingKey routingkey
 */
async bindExchange(srcExchangeName : string, routingKey : string)

unbindExchange

/**
 * 解除绑定
 * @param srcExchangeName exchange名
 * @param routingKey routingKey
 */
async unbindExchange(srcExchangeName : string, routingKey : string)

bindHeaders

/**
 * 绑定Header
 * @param exchangeName exchange名
 * @param routingKey routingKey
 */
async bindHeaders(exchangeName : string, routingKey : string)

getExchagneInstance

/**
 * 获取Exchange的实例
 */
getExchagneInstance()

isInit

/**
 * 是否初始化
 */
isInit()

getName

/**
 * 获取Exchange名
 */
getName()

C4Queue

说明：对应RabbitMQ的Queue
路径：./src/C4Queue.ts
成员变量：
- m_Channel，AMQPClient，MQ连接
- m_Queue，AMQPQueue
- m_Name，名字
- m_ConsumerTag，自定义标签
- m_MsgQueue，用于预取消息的队列
- m_Logger，日志对象(可使用console，推荐使用c4logger)

成员方法：

declared

/**
 * 声明
 * @param Conn 连接
 * @param name 名字
 * @param options QueueOptions
 * @param logger 日志
 */
async declared(Conn : C4MQ, name : string, options : QueueOptions, logger ?: any)

desroy

/**
 * 销毁
 * @param options {
 *  ifUnused : boolean,     // 是否在不用后销毁
 *  isEmpty : boolean,      // 是否在空后后销毁
 * }
 */
desroy(options : { ifUnused : boolean, isEmpty : boolean})

bindExchange

/**
 * 绑定Excahnge
 * @param exchangeName excahnge的名字
 * @param routingKey routingKey
 */
async bindExchange(exchangeName : string, routingKey : string)

unbindExchange

/**
 * 解绑
 * @param exchangeName exchange的名字
 * @param routingKey routingKey
 */
unbindExchange(exchangeName : string, routingKey : string)

bindHeaders

/**
 * 绑定Headers
 * @param exchangeName excahnge名字
 * @param routingKey routingKey
 */
bindHeaders(exchangeName : string, routingKey : string)

unbindHeaders

/**
 * 解绑Headers
 * @param exchangeName exchange名
 * @param routingKey routingKeys
 */
unbindHeaders(exchangeName : string, routingKey : string)

/**
 * 订阅消息
 * @param options 订阅配置
 * @param CB 消息处理方法，如果不设置，则采用内部机制
 * 
 */
async subscribe(options : SubscribeOptions, CB ?: SubscribeCallback)

unsubscribe

/**
 * 取消订阅
 */
unsubscribe()

peekMsg

/**
 * 去消息并处理
 * @param CB 消息处理方法
 * 订阅消息时没有设置处理方法，则会进行内部处理，然后使用该方法进行单个消息的处理
 * CB返回true，则消息正常ack
 * CB返回false，则消息丢弃不会重排
 * CB抛异常，则消息重排
 */
async peekMsg(CB : (message : any,
    headers : {[key : string] : any},
    deliveryInfo : DeliveryInfo,
    ack : Ack) => Promise<boolean> | boolean)

getQueueInstance

/**
 * 获取Queue实例
 */
getQueueInstance()

isInit

/**
 * 是否已经初始化
 */
isInit()

getName

/**
 * 获取Queue名
 */
getName()

C4Publisher

说明：对C4Exchange的封装，主要发送消息；
路径：./src/C4Publisher.ts
成员变量：
- m_Exchange，C4Exchange；
- m_PublishOption，ExchangePublishOptions；
- m_RoutingKey，routingkey；
- m_Logger，日志对象

成员方法：

init

/**
 * 初始化
 * @param Conn C4MQ，MQ连接
 * @param option C4PublisherOption
 * @param logger 日志对象
 */
async init(Conn : C4MQ, option : C4PublisherOption, logger ?: any)

publish

/**
 * 发布消息
 * @param message 消息
 * @param routingKey routingKey
 * @param option ExchangePublishOptions
 */
 async publish(message : {} | Buffer, routingKey ?: string, option ?: ExchangePublishOptions)

isInit

/**
 * 是否初始化
 */
isInit()

getName

/**
 * 获取Exchange的名字
 */
getName()

C4Subscriber

说明：对C4Queue的封装，支持注解添加消息处理方法；
路径：./src/C4Subscriber.ts
成员变量：
- m_Queue，C4Queue；
- m_DefaultPublisher，默认绑定的Publisher名；
- m_SubscribeOptions，订阅配置项；
- m_Logger，日志对象；
- m_CBs，消息处理方法；

成员方法：

init

/**
 * 初始化
 * @param Conn C4MQ连接，MQ连接
 * @param option C4SubscriberOption
 * @param logger 日志对象
 */
async init(Conn : C4MQ, option : C4SubscriberOption, logger ?: any)

addSubscribe

/**
 * 添加订阅配置
 * @param option C4SubscribeOption
 */
addSubscribe(option : C4SubscribeOption)

/**
 * 开始订阅
 */
async subscribe()

__processMsg

/**
 * 消息处理方法
 */
async __processMsg()

addMQHandler

/**
 * 添加消息处理方法
 * @param handlers 方法对象或者加名字
 * 当handlers为名字时，会从
 * './MQHandlers/',
 * './out/MQHandlers/'
 * 两个位置尝试进行搜索对应的文件并加载
 * ts：默认放到在src/MQHandlers目下编写ts代码，
 * 编译后在out目录下自动得到'./out/MQHandlers/'
 * js：工程目录下创建MQHandlers，然后放入编译后的js代码
 */
async addMQHandler(handlers : Array<any>) : Promise<any>;
async addMQHandler(handlerPaths : string[]) : Promise<any>;
async addMQHandler(arg : Array<any | string>)

isInit

/**
 * 是否初始化
 */
isInit()

getName

/**
 * 获取Queue的名字
 */
getName()

Keywords

C4Framework

FAQs

What is c4-mq?

Is c4-mq popular?

Is c4-mq well maintained?

Package last updated on 26 Oct 2018

Did you know?

Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.

Install