Redis adapter about moleculer-channels HOT 9 CLOSED

// static-service.js
const { ServiceBroker } = require("moleculer");
const ChannelsMiddleware = require("@moleculer/channels").Middleware;
const broker = new ServiceBroker({
  namespace: "test",
  nodeID: "test1",
  transporter: "TCP",
  middlewares:[ChannelsMiddleware({
    adapter:"redis://127.0.0.1:6379"
  })]
});

// Start the Moleculer broker
broker.start();

broker.repl();

setInterval(() => {
  broker.sendToChannel("order.created", {
    id: 1234,
    items: "test"
  });
}, 100);


// Start the service to serve static resources

const serviceSchema = {
  name: "subscriber",
  channels: {
    "order.created": {
      group: "mygroup",
      redis: {
        minIdleTime: 1000,
        claimInterval: 1,
        startID: "0"
      },
      maxRetries: 100,
      handler(payload) {
          console.log(payload);
          throw new Error;
      }
    }
  },
}
broker.createService(serviceSchema).then();

// send 10 times

broker.destroyService("subscriber").then(()=>{
  setTimeout(()=>{
    // then create new Service, Remaining transmission times data loss

broker.createService(serviceSchema);
  }, 10000);
});

const pubClient = this.clients.get(this.pubName);
// 1. Delete consumer from the consumer group
// 2. Do NOT destroy the consumer group
// https://redis.io/commands/XGROUP
return pubClient.xgroup(
	"DELCONSUMER",
	chan.name, // Stream Name
	chan.group, // Consumer Group
	chan.id // Consumer ID
);

const pubClient = this.clients.get(this.pubName);
let pending = await pubClient.xpending(
	chan.name,
	chan.group,
	"-",
	"+",
	10 // Max reported entries
);
if(!pending) {
	// 1. Delete consumer from the consumer group
	// 2. Do NOT destroy the consumer group
	// https://redis.io/commands/XGROUP
	return pubClient.xgroup(
		"DELCONSUMER",
		chan.name, // Stream Name
		chan.group, // Consumer Group
		chan.id // Consumer ID
	);
}