egg-pulsar

Apache pulsar plugin for egg

中文文档

Install

// https://pulsar.apache.org/docs/en/client-libraries-cpp/

// install pulsar-client-dev lib for macOS
$ brew install libpulsar

$ npm i egg-pulsar --save

Usage

// {app_root}/config/plugin.js
exports.pulsar = {
  enable: true,
  package: 'egg-pulsar',
};

Configuration

// {app_root}/config/config.default.js
exports.pulsar = {
  client: {
    url: 'pulsar://localhost:6650',
    options: {
      operationTimeoutSeconds: 30,
    },
    subscribe: {
      topic: 'persistent://public/default/my-topic',
      subscription: 'sub1',
      subscriptionType: 'Shared',
      ackTimeoutMs: 10000,
      listener: 'xxx', // 对应 app/subscriber/xxx
    },
  }
};

see apache pulsar client-libraries-node for more detail.

Example

Producer Send Message

// {app_root}/app/controller/home.js
'use strict';

const Controller = require('egg').Controller;

class HomeController extends Controller {
  async index() {
    const { ctx } = this;

    await this.app.pulsar.send({ now: Date.now() }, 'my-topic');
    // or
    await this.app.pulsar.send({ now: Date.now() }, { topic: 'my-topic' });
    // or
    await this.app.pulsar.send({ data: { now: Date.now() }, properties: [ 'a', 'c' ] }, { topic: 'my-topic' });

    ctx.body = 'hi, egg';
  }
}

module.exports = HomeController;

Consumer Message Of Subscriber

// {app_root}/app/subscriber/test.js
'use strict';

const debug = require('debug')('subscriber');

class TestSubscriber {
  constructor(ctx) {
    this.ctx = ctx;
    this.app = ctx.app;
  }

  async consume(message) {
    debug('message: %s', JSON.stringify(message));
    console.log(message);
    // throw ('consume message error');
  }
}

module.exports = TestSubscriber;

Test

// pulsar server run on docker 
$ docker run -it \
-p 6650:6650 \
-p 8080:8080 \
--mount source=pulsardata,target=/pulsar/data \
--mount source=pulsarconf,target=/pulsar/conf \
apachepulsar/pulsar:2.7.2 \
bin/pulsar standalone

Questions & Suggestions

Please open an issue here.

License

MIT