egg-redis-stream

Redis Stream message queue.

Install

$ npm i egg-redis-stream egg-redis --save

Usage

This module dependent on egg-redis plugin, so we must enable both.

// {app_root}/config/plugin.ts

const plugin: EggPlugin = {
  redis: {
    enable: true,
    package: 'egg-redis',
  },
  redisStream: {
    enable: true,
    package: 'egg-redis-stream',
  },
};

export default plugin;

Configuration

// {app_root}/config/config.default.ts

export default (appInfo: EggAppInfo) => {
  const config = {} as PowerPartial<EggAppConfig>;

  // other config...

  config.producer = {
    // {app_root}/producer/xxxx.ts
    xxxx: 'topic name',
  };

  config.consumer = {
    // {app_root}/consumer/xxxx.ts
    xxxx: {
      topic: 'topic name',
      group: 'consumer group',
      name: 'consumer name',
    },
  };

  config.redis = {
    client: {
      port: 6379,
      host: '127.0.0.1',
      password: '123456',
    },
  };

  // the return config will combines to EggAppConfig
  return config;
};

Example

1. Configure producer/consumer.

// {app_root}/config/config.default.ts

config.consumer = {
  // {app_root}/consumer/order.ts
  order: {
    topic: 'order_topic',
    group: 'default',
    name: 'local',
  },
};

config.producer = {
  // {app_root}/producer/order.ts
  order: 'order_topic',
};

2. Create producer

// {app_root}/producer/order.ts

export interface OrderItem {
  id: string;
  record: string;
}

export default class OrderProducer extends Producer<OrderItem> {
  // You can create some function
}

3. Produce a message

// {app_root}/service/someService.ts

export default class SomeService extends Service {
  public async xxxx() {
    await app.producer.order.push({
      id: 'ORD5**********W9JJxokx',
      record: '...',
    });
  }
}

4. Create consumer

// {app_root}/consumer/order.ts

export interface Order {
  id: string;
  record: string;
}

export default class OrderConsumer extends Consumer<Order> {
  protected async process(id: string, message: Order) {
    const { ctx } = this;
    ctx.logger.info(message);
    await this.ack(id);
    // or app.consumer.order.ack(id);
  }
}

Questions & Suggestions

Please open an issue here.

License

MIT