1+ import { expect } from "chai" ;
2+ import Redis , { Cluster } from "../../lib" ;
3+ import redis from "../../lib" ;
4+
5+ const host = "127.0.0.1" ;
6+ const masters = [ 30000 , 30001 , 30002 ] ;
7+ const port : number = masters [ 0 ]
8+
9+ /**
10+ * Wait for a specified time
11+ *
12+ * @param ms
13+ */
14+ function sleep ( ms : number ) {
15+ return new Promise ( resolve => setTimeout ( resolve , ms ) ) ;
16+ }
17+
18+
19+ describe ( "cluster:ClusterSubscriberGroup" , ( ) => {
20+
21+ it ( "should unsubscribe from the given channel" , async ( ) => {
22+
23+ const cluster : Cluster = new Cluster ( [ { host : host , port : port } ] , { shardedSubscribers : true } ) ;
24+
25+ //Subscribe to the three channels
26+ cluster . ssubscribe ( "channel:1:{1}" , "channel:2:{1}" , "channel:3:{1}" ) . then ( ( count : number ) => {
27+ console . log ( "Subscribed to 3 channels." ) ;
28+ expect ( count ) . to . equal ( 3 ) ;
29+ } ) ;
30+
31+ //Publish a message to one of the channels
32+ cluster . spublish ( "channel:2:{1}" , "This is a test message to channel 2." ) . then ( ( value : number ) => {
33+ console . log ( "Published a message to channel:2:{1} and expect one subscriber." ) ;
34+ expect ( value ) . to . be . eql ( 1 ) ;
35+ } ) ;
36+
37+ await sleep ( 500 ) ;
38+
39+ //Unsubscribe from one of the channels
40+ cluster . sunsubscribe ( "channel:2:{1}" ) . then ( ( count : number ) => {
41+ console . log ( "Unsubscribed from channel:2:{1}." ) ;
42+ expect ( count ) . to . equal ( 2 ) ;
43+ } ) ;
44+
45+ await sleep ( 500 ) ;
46+
47+ //Publish a message to the channel from which we unsubscribed
48+ cluster . spublish ( "channel:2:{1}" , "This is a test message to channel 2." ) . then ( ( value : number ) => {
49+ console . log ( "Published a second message to channel:2:{1} and expect to have nobody listening." ) ;
50+ expect ( value ) . to . be . eql ( 0 ) ;
51+ } ) ;
52+
53+ await sleep ( 1000 ) ;
54+ await cluster . disconnect ( ) ;
55+ } ) ;
56+
57+ it ( "works when ssubscribe only works for keys that map to the same slot" , async ( ) => {
58+
59+ const cluster : Cluster = new Cluster ( [ { host : host , port : port } ] , { shardedSubscribers : true } ) ;
60+
61+ //Register the callback
62+ cluster . on ( "smessage" , ( channel , message ) => {
63+ console . log ( message ) ;
64+ expect ( message . startsWith ( "This is a test message" ) ) . to . be . true ;
65+ } ) ;
66+
67+ //Subscribe to the channels on different slots
68+ cluster . ssubscribe ( "channel{my}:1" , "channel{yours}:2" ) . then ( ( count : number ) => {
69+ //Should not be called
70+ expect ( true ) . to . equal ( false ) ;
71+ } ) . catch ( ( err ) => {
72+ expect ( err . toString ( ) . conaints ( "CROSSSLOT Keys in request don't hash to the same slot" ) ) . to . be . true ;
73+ } ) ;
74+
75+ //Subscribe to the channels on the same slot
76+ cluster . ssubscribe ( "channel{my}:1" , "channel{my}:2" ) . then ( ( count : number ) => {
77+ console . log ( count ) ;
78+ expect ( count ) . to . equal ( 2 ) ;
79+ } ) . catch ( ( err ) => {
80+ expect ( true ) . to . equal ( false ) ;
81+ } ) ;
82+
83+ //Subscribe once again on the other slot
84+ cluster . ssubscribe ( "channel{yours}:2" ) . then ( ( count : number ) => {
85+ console . log ( count ) ;
86+ expect ( count ) . to . equal ( 1 ) ;
87+ } ) . catch ( ( err ) => {
88+ expect ( true ) . to . equal ( false ) ;
89+ } ) ;
90+
91+ //Publish messages
92+ cluster . spublish ( "channel{my}:1" , "This is a test message to my first channel." ) . then ( ( value : number ) => {
93+ console . log ( "Published a message to channel{my}:1" ) ;
94+ expect ( value ) . to . be . eql ( 1 ) ;
95+ } ) ;
96+
97+ cluster . spublish ( "channel{my}:2" , "This is a test message to my second channel." ) . then ( ( value : number ) => {
98+ console . log ( "Published a message to channel{my}:2" ) ;
99+ expect ( value ) . to . be . eql ( 1 ) ;
100+ } ) ;
101+
102+ cluster . spublish ( "channel{yours}:2" , "This is a test message to your second channel." ) . then ( ( value : number ) => {
103+ console . log ( "Published a message to channel{yours}:2" ) ;
104+ expect ( value ) . to . be . eql ( 1 ) ;
105+ } ) ;
106+
107+ //Give it some time to process messages and then disconnect
108+ await sleep ( 1000 ) ;
109+ await cluster . disconnect ( ) ;
110+ } ) ;
111+
112+
113+ it ( "works when you can receive published messages to all primary nodes after having subscribed" , async ( ) => {
114+
115+ // 0. Prepare the publisher and the subscriber
116+ const publisher : Cluster = new Cluster ( [ { host : host , port : port } ] ) ;
117+
118+ //-- Publisher
119+ // Verify that the cluster is configured with 3 master nodes
120+ publisher . on ( "ready" , ( ) => {
121+ expect ( publisher . nodes ( "master" ) . length ) . to . eql ( 3 ) ;
122+ } ) ;
123+
124+ //-- Subscriber
125+ const subscriber : Cluster = new Cluster ( [ { host : host , port : port } ] , { shardedSubscribers : true } ) ;
126+ let totalNumMessages = 0 ;
127+
128+ // Register the subscriber callback
129+ subscriber . on ( "smessage" , ( channel , message ) => {
130+ console . log ( message ) ;
131+ expect ( message . startsWith ( "This is a test message" ) ) . to . eql ( true ) ;
132+ expect ( message . endsWith ( channel + "." ) ) . to . eql ( true ) ;
133+ totalNumMessages ++ ;
134+ expect ( totalNumMessages ) . to . lte ( 3 ) ;
135+ } ) ;
136+
137+ //Verify that we did not get more than 3 subscribers
138+ let numSubs = 0 ;
139+ subscriber . on ( "+subscriber" , ( ) => {
140+ numSubs ++
141+ expect ( numSubs ) . to . lte ( 3 ) ;
142+ } ) ;
143+
144+ //1. Construct 3 channel names, whereby the first one is expected to land on node 1, the second one on node 2, and so on
145+ const channels = [ "channel:test:3" , "channel:test:2" , "channel:test:0" ]
146+
147+ for ( const c of channels ) {
148+ console . log ( "Trying to publish to channel:" , c ) ;
149+
150+ //2. Subscribe to the channel
151+ await subscriber . ssubscribe ( c )
152+
153+ //3. Publish a message before initializing the message handling
154+ const numSubscribers = await publisher . spublish ( c , "This is a test message " + c + "." ) ;
155+ expect ( numSubscribers ) . to . eql ( 1 ) ;
156+ }
157+
158+ //Give it some time to process messages and then disconnect
159+ await sleep ( 1000 ) ;
160+ subscriber . disconnect ( ) ;
161+ } ) ;
162+
163+ it ( "receive messages on the channel after the slot was moved" , async ( ) => {
164+
165+ //The hash slot of interest
166+ const slot = 2318 ;
167+ const channel = "channel:test:3" ;
168+
169+ //Used as control connections for orchestrating the slot migration
170+ const source : Redis = new Redis ( { host : host , port : 30000 } ) ;
171+ const target : Redis = new Redis ( { host : host , port : 30001 } ) ;
172+
173+ //Initialize the publisher cluster connections and verify that the slot is on node 1
174+ const publisher : Cluster = new Cluster ( [ { host : host , port : port } ] ) ;
175+
176+ publisher . on ( "ready" , ( ) => {
177+ expect ( publisher . slots [ slot ] [ 0 ] ) . eql ( "127.0.0.1:30000" ) ;
178+ } ) ;
179+
180+
181+ //Initialize the subscriber cluster connections and verify that the slot is on node 1
182+ const subscriber : Cluster = new Cluster ( [ { host : host , port : port } ] , { shardedSubscribers : true } ) ;
183+
184+ subscriber . on ( "ready" , ( ) => {
185+ expect ( subscriber . slots [ slot ] [ 0 ] ) . eql ( "127.0.0.1:30000" )
186+ } ) ;
187+
188+ //The subscription callback. We should receive both messages
189+ let totalNumMessages = 0 ;
190+ subscriber . on ( "smessage" , ( channel , message ) => {
191+ totalNumMessages ++ ;
192+
193+ if ( totalNumMessages == 1 ) {
194+ console . log ( "Received the first message" ) ;
195+ expect ( message . includes ( "#1" ) ) . to . eql ( true ) ;
196+ }
197+
198+ if ( totalNumMessages == 2 ) {
199+ console . log ( "Received the second message" ) ;
200+ expect ( message . includes ( "#2" ) ) . to . eql ( true ) ;
201+ }
202+ } ) ;
203+
204+ //Subscribe and then publish
205+ await subscriber . ssubscribe ( channel ) ;
206+ await publisher . spublish ( channel , "This is a test message #1 to slot "
207+ + slot + " on channel " + channel + "." ) ;
208+
209+ //Get the target node
210+ const nodes = await source . cluster ( 'SLOTS' ) ;
211+ const sourceNode = nodes [ 0 ] [ 2 ] [ 2 ] ;
212+ const targetNode = nodes [ 1 ] [ 2 ] [ 2 ] ;
213+
214+ //Migrate the slot
215+ console . log ( `Migrating slot ${ slot } ${ targetNode } ) ;
216+ let status = ""
217+ status = await target . cluster ( "SETSLOT" , slot , "IMPORTING" , targetNode ) ;
218+ expect ( status ) . to . eql ( "OK" ) ;
219+ status = await source . cluster ( 'SETSLOT' , slot , 'MIGRATING' , sourceNode ) ;
220+ expect ( status ) . to . eql ( "OK" ) ;
221+ status = await target . cluster ( "SETSLOT" , slot , "NODE" , targetNode ) ;
222+ expect ( status ) . to . eql ( "OK" ) ;
223+ status = await source . cluster ( "SETSLOT" , slot , "NODE" , targetNode ) ;
224+ expect ( status ) . to . eql ( "OK" ) ;
225+
226+ //Trigger a topology update on the subscriber. This needs at least one moved response.
227+ //TODO: What if there is no traffic on the cluster connection?
228+ status = await subscriber . set ( "match_slot{" + channel + "}" , "channel 3" ) ;
229+ expect ( status ) . to . eql ( "OK" ) ;
230+ expect ( subscriber . slots [ slot ] [ 0 ] ) . eql ( "127.0.0.1:30001" ) ;
231+
232+ //Wait a bit to let the subscriber resubscribe to previous channels
233+ await sleep ( 1000 ) ;
234+
235+ const numSubscribers = await publisher . spublish ( channel , "This is a test message #2 to slot "
236+ + slot + " on channel " + channel + "." ) ;
237+ expect ( publisher . slots [ slot ] [ 0 ] ) . eql ( "127.0.0.1:30001" ) ;
238+ expect ( numSubscribers ) . to . eql ( 1 ) ;
239+ } ) ;
240+ } ) ;
0 commit comments