4007873007

Build Type

push

github

Committed by GitHub

Commit Message

fix: unsubscribe not work with stringNumbers (#1710)

Run Details

1136 of 1304 branches covered (87.12%)

Branch coverage included in aggregate %.

4 of 4 new or added lines in 1 file covered. (100.0%)

19 existing lines in 5 files now uncovered.

2168 of 2314 relevant lines covered (93.69%)

3951.65 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.98

/lib/cluster/ClusterSubscriber.ts

import { EventEmitter } from "events";
import ConnectionPool from "./ConnectionPool";
import { getConnectionName, getNodeKey } from "./util";
import { sample, noop, Debug } from "../utils";
import Redis from "../Redis";

const debug = Debug("cluster:subscriber");

export default class ClusterSubscriber {
  private started = false;
  private subscriber: any = null;
  private lastActiveSubscriber: any;

  constructor(
    private connectionPool: ConnectionPool,
    private emitter: EventEmitter
  ) {
    // If the current node we're using as the subscriber disappears
    // from the node pool for some reason, we will select a new one
    // to connect to.
    // Note that this event is only triggered if the connection to
    // the node has been used; cluster subscriptions are setup with
    // lazyConnect = true. It's possible for the subscriber node to
    // disappear without this method being called!
    // See https://github.com/luin/ioredis/pull/1589
    this.connectionPool.on("-node", (_, key: string) => {
      if (!this.started || !this.subscriber) {
        return;
      }
      if (getNodeKey(this.subscriber.options) === key) {
        debug("subscriber has left, selecting a new one...");
        this.selectSubscriber();
      }
    });
    this.connectionPool.on("+node", () => {
      if (!this.started || this.subscriber) {
        return;
      }
      debug(
        "a new node is discovered and there is no subscriber, selecting a new one..."
      );
      this.selectSubscriber();
    });
  }

  getInstance(): any {
    return this.subscriber;
  }

  start(): void {
    this.started = true;
    this.selectSubscriber();
    debug("started");
  }

  stop(): void {
    this.started = false;
    if (this.subscriber) {
      this.subscriber.disconnect();
      this.subscriber = null;
    }
    debug("stopped");
  }

  private onSubscriberEnd = () => {
    if (!this.started) {
      debug(
        "subscriber has disconnected, but ClusterSubscriber is not started, so not reconnecting."
      );
      return;
    }
    // If the subscriber closes whilst it's still the active connection,
    // we might as well try to connecting to a new node if possible to
    // minimise the number of missed publishes.
    debug("subscriber has disconnected, selecting a new one...");
    this.selectSubscriber();
  };

  private selectSubscriber() {
    const lastActiveSubscriber = this.lastActiveSubscriber;

    // Disconnect the previous subscriber even if there
    // will not be a new one.
    if (lastActiveSubscriber) {
      lastActiveSubscriber.off("end", this.onSubscriberEnd);
      lastActiveSubscriber.disconnect();
    }

    if (this.subscriber) {
      this.subscriber.off("end", this.onSubscriberEnd);
      this.subscriber.disconnect();
    }

    const sampleNode = sample(this.connectionPool.getNodes());
    if (!sampleNode) {
      debug(
        "selecting subscriber failed since there is no node discovered in the cluster yet"
      );
      this.subscriber = null;
      return;
    }

    const { options } = sampleNode;
    debug("selected a subscriber %s:%s", options.host, options.port);

    /*
     * Create a specialized Redis connection for the subscription.
     * Note that auto reconnection is enabled here.
     *
     * `enableReadyCheck` is also enabled because although subscription is allowed
     * while redis is loading data from the disk, we can check if the password
     * provided for the subscriber is correct, and if not, the current subscriber
     * will be disconnected and a new subscriber will be selected.
     */
    this.subscriber = new Redis({
      port: options.port,
      host: options.host,
      username: options.username,
      password: options.password,
      enableReadyCheck: true,
      connectionName: getConnectionName("subscriber", options.connectionName),
      lazyConnect: true,
      tls: options.tls,
      // Don't try to reconnect the subscriber connection. If the connection fails
      // we will get an end event (handled below), at which point we'll pick a new
      // node from the pool and try to connect to that as the subscriber connection.
      retryStrategy: null,
    });

    // Ignore the errors since they're handled in the connection pool.
    this.subscriber.on("error", noop);

    // The node we lost connection to may not come back up in a
    // reasonable amount of time (e.g. a slave that's taken down
    // for maintainence), we could potentially miss many published
    // messages so we should reconnect as quickly as possible, to
    // a different node if needed.
    this.subscriber.once("end", this.onSubscriberEnd);

    // Re-subscribe previous channels
    const previousChannels = { subscribe: [], psubscribe: [], ssubscribe: [] };
    if (lastActiveSubscriber) {
      const condition =
        lastActiveSubscriber.condition || lastActiveSubscriber.prevCondition;
      if (condition && condition.subscriber) {
        previousChannels.subscribe = condition.subscriber.channels("subscribe");
        previousChannels.psubscribe =
          condition.subscriber.channels("psubscribe");
        previousChannels.ssubscribe =
          condition.subscriber.channels("ssubscribe");
      }
    }
    if (
      previousChannels.subscribe.length ||
      previousChannels.psubscribe.length ||
      previousChannels.ssubscribe.length
    ) {
      let pending = 0;
      for (const type of ["subscribe", "psubscribe", "ssubscribe"]) {
        const channels = previousChannels[type];
        if (channels.length) {
          pending += 1;
          debug("%s %d channels", type, channels.length);
          this.subscriber[type](channels)
            .then(() => {
              if (!--pending) {
                this.lastActiveSubscriber = this.subscriber;
              }
            })
            .catch(() => {
              // TODO: should probably disconnect the subscriber and try again.
              debug("failed to %s %d channels", type, channels.length);
            });
        }
      }
    } else {
      this.lastActiveSubscriber = this.subscriber;
    }
    for (const event of [
      "message",
      "messageBuffer",
      "smessage",
      "smessageBuffer",
    ]) {
      this.subscriber.on(event, (arg1, arg2) => {
        this.emitter.emit(event, arg1, arg2);
      });
    }
    for (const event of ["pmessage", "pmessageBuffer"]) {
      this.subscriber.on(event, (arg1, arg2, arg3) => {
        this.emitter.emit(event, arg1, arg2, arg3);
      });
    }
  }
}

1	import { EventEmitter } from "events";
2	import ConnectionPool from "./ConnectionPool";
3	import { getConnectionName, getNodeKey } from "./util";	1✔
4	import { sample, noop, Debug } from "../utils";	1✔
5	import Redis from "../Redis";	1✔
6
7	const debug = Debug("cluster:subscriber");	1✔
8
9	export default class ClusterSubscriber {	1✔
10	private started = false;	110✔
11	private subscriber: any = null;	110✔
12	private lastActiveSubscriber: any;
13
14	constructor(
15	private connectionPool: ConnectionPool,	110✔
16	private emitter: EventEmitter	110✔
17	) {
18	// If the current node we're using as the subscriber disappears
19	// from the node pool for some reason, we will select a new one
20	// to connect to.
21	// Note that this event is only triggered if the connection to
22	// the node has been used; cluster subscriptions are setup with
23	// lazyConnect = true. It's possible for the subscriber node to
24	// disappear without this method being called!
25	// See https://github.com/luin/ioredis/pull/1589
26	this.connectionPool.on("-node", (_, key: string) => {	110✔
27	if (!this.started \|\| !this.subscriber) {	227✔
28	return;	208✔
29	}
30	if (getNodeKey(this.subscriber.options) === key) {	19✔
31	debug("subscriber has left, selecting a new one...");	15✔
32	this.selectSubscriber();	15✔
33	}
34	});
35	this.connectionPool.on("+node", () => {	110✔
36	if (!this.started \|\| this.subscriber) {	227✔
37	return;	222✔
38	}
39	debug(	5✔
40	"a new node is discovered and there is no subscriber, selecting a new one..."
41	);
42	this.selectSubscriber();	5✔
43	});
44	}
45
46	getInstance(): any {
47	return this.subscriber;	14✔
48	}
49
50	start(): void {
51	this.started = true;	113✔
52	this.selectSubscriber();	113✔
53	debug("started");	110✔
54	}
55
56	stop(): void {
57	this.started = false;	114✔
58	if (this.subscriber) {	114✔
59	this.subscriber.disconnect();	97✔
60	this.subscriber = null;	97✔
61	}
62	debug("stopped");	114✔
63	}
64
65	private onSubscriberEnd = () => {	110✔
66	if (!this.started) {	94!
67	debug(	94✔
68	"subscriber has disconnected, but ClusterSubscriber is not started, so not reconnecting."
69	);
70	return;	94✔
71	}
72	// If the subscriber closes whilst it's still the active connection,
73	// we might as well try to connecting to a new node if possible to
74	// minimise the number of missed publishes.
UNCOV 75	debug("subscriber has disconnected, selecting a new one...");	×
UNCOV 76	this.selectSubscriber();	×
77	};
78
79	private selectSubscriber() {
80	const lastActiveSubscriber = this.lastActiveSubscriber;	133✔
81
82	// Disconnect the previous subscriber even if there
83	// will not be a new one.
84	if (lastActiveSubscriber) {	133✔
85	lastActiveSubscriber.off("end", this.onSubscriberEnd);	36✔
86	lastActiveSubscriber.disconnect();	36✔
87	}
88
89	if (this.subscriber) {	133✔
90	this.subscriber.off("end", this.onSubscriberEnd);	22✔
91	this.subscriber.disconnect();	22✔
92	}
93
94	const sampleNode = sample(this.connectionPool.getNodes());	133✔
95	if (!sampleNode) {	133✔
96	debug(	14✔
97	"selecting subscriber failed since there is no node discovered in the cluster yet"
98	);
99	this.subscriber = null;	14✔
100	return;	14✔
101	}
102
103	const { options } = sampleNode;	119✔
104	debug("selected a subscriber %s:%s", options.host, options.port);	119✔
105
106	/*
107	* Create a specialized Redis connection for the subscription.
108	* Note that auto reconnection is enabled here.
109	*
110	* `enableReadyCheck` is also enabled because although subscription is allowed
111	* while redis is loading data from the disk, we can check if the password
112	* provided for the subscriber is correct, and if not, the current subscriber
113	* will be disconnected and a new subscriber will be selected.
114	*/
115	this.subscriber = new Redis({	119✔
116	port: options.port,
117	host: options.host,
118	username: options.username,
119	password: options.password,
120	enableReadyCheck: true,
121	connectionName: getConnectionName("subscriber", options.connectionName),
122	lazyConnect: true,
123	tls: options.tls,
124	// Don't try to reconnect the subscriber connection. If the connection fails
125	// we will get an end event (handled below), at which point we'll pick a new
126	// node from the pool and try to connect to that as the subscriber connection.
127	retryStrategy: null,
128	});
129
130	// Ignore the errors since they're handled in the connection pool.
131	this.subscriber.on("error", noop);	119✔
132
133	// The node we lost connection to may not come back up in a
134	// reasonable amount of time (e.g. a slave that's taken down
135	// for maintainence), we could potentially miss many published
136	// messages so we should reconnect as quickly as possible, to
137	// a different node if needed.
138	this.subscriber.once("end", this.onSubscriberEnd);	119✔
139
140	// Re-subscribe previous channels
141	const previousChannels = { subscribe: [], psubscribe: [], ssubscribe: [] };	119✔
142	if (lastActiveSubscriber) {	119✔
143	const condition =
144	lastActiveSubscriber.condition \|\| lastActiveSubscriber.prevCondition;	22✔
145	if (condition && condition.subscriber) {	22✔
146	previousChannels.subscribe = condition.subscriber.channels("subscribe");	5✔
147	previousChannels.psubscribe =	5✔
148	condition.subscriber.channels("psubscribe");
149	previousChannels.ssubscribe =	5✔
150	condition.subscriber.channels("ssubscribe");
151	}
152	}
153	if (	119✔
154	previousChannels.subscribe.length \|\|	350✔
155	previousChannels.psubscribe.length \|\|
156	previousChannels.ssubscribe.length
157	) {
158	let pending = 0;	5✔
159	for (const type of ["subscribe", "psubscribe", "ssubscribe"]) {	5✔
160	const channels = previousChannels[type];	12✔
161	if (channels.length) {	12✔
162	pending += 1;	5✔
163	debug("%s %d channels", type, channels.length);	5✔
164	this.subscriber[type](channels)	5✔
165	.then(() => {
UNCOV 166	if (!--pending) {	×
UNCOV 167	this.lastActiveSubscriber = this.subscriber;	×
168	}
169	})
170	.catch(() => {
171	// TODO: should probably disconnect the subscriber and try again.
172	debug("failed to %s %d channels", type, channels.length);	2✔
173	});
174	}
175	}
176	} else {
177	this.lastActiveSubscriber = this.subscriber;	114✔
178	}
179	for (const event of [	116✔
180	"message",
181	"messageBuffer",
182	"smessage",
183	"smessageBuffer",
184	]) {
185	this.subscriber.on(event, (arg1, arg2) => {	464✔
186	this.emitter.emit(event, arg1, arg2);	4✔
187	});
188	}
189	for (const event of ["pmessage", "pmessageBuffer"]) {	116✔
190	this.subscriber.on(event, (arg1, arg2, arg3) => {	232✔
UNCOV 191	this.emitter.emit(event, arg1, arg2, arg3);	×
192	});
193	}
194	}
195	}

luin / ioredis / 4007873007

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous