-
Notifications
You must be signed in to change notification settings - Fork 17
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #23 from lujiajing1126/blocking
Support Blocking Mode
- Loading branch information
Showing
5 changed files
with
184 additions
and
63 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,116 @@ | ||
| const colors = require('colors'); | ||
| const util = require('util'); | ||
|
|
||
| const INT_PREFIX = "(integer)"; | ||
| const BLOCKING_CMDS = ["subscribe", "monitor", "psubscribe"]; | ||
|
|
||
| class Executor { | ||
|
|
||
| constructor(client, commands) { | ||
| this._client = client; | ||
| this.commands = commands; | ||
|
|
||
| const CMD = this.commands.shift().toLowerCase(); | ||
| this.blockingMode = BLOCKING_CMDS.includes(CMD); | ||
|
|
||
| this._executor = this._client.client[`${CMD}Async`]; | ||
|
|
||
| if (typeof this._executor !== "function") { | ||
| this._executor = this._client.client[`send_commandAsync`]; | ||
| // recombine commands | ||
| this.commands = [CMD, this.commands]; | ||
| } | ||
| } | ||
|
|
||
| writeResult(result) { | ||
| if (Array.isArray(result)) { | ||
| this._client.next = result.map((item, index) => { | ||
| return util.format("%d) %s", index + 1, item); | ||
| }); | ||
| } else if (result === null) { | ||
| this._client.next = "(nil)"; | ||
| } else if (typeof result === 'object') { | ||
| this._client.next = Object.entries(result).flat().map((item, index) => { | ||
| return util.format("%d) %s", index + 1, item); | ||
| }); | ||
| } else { | ||
| // number or string | ||
| // default to print it as `string` | ||
| this._client.next = util.format(Number.isInteger(result) ? `${INT_PREFIX} ${result}` : result); | ||
| } | ||
| } | ||
|
|
||
| run() { | ||
| return this._executor.bind(this._client.client)(...this.commands) | ||
| .then((result) => { | ||
| this.writeResult(result); | ||
| return this.blockingMode; | ||
| }).catch((e) => { | ||
| this._client.next = colors.red(`(error) ${e.message}`); | ||
| }); | ||
| } | ||
|
|
||
| shutdown() { | ||
| // do nothing | ||
| } | ||
| } | ||
|
|
||
| class SubscribeExecutor extends Executor { | ||
| constructor(client, commands) { | ||
| super(client, commands); | ||
| } | ||
|
|
||
| run() { | ||
| this._client.client.on("subscribe", (channel, count) => {}); | ||
|
|
||
| this._client.client.on("message", (channel, message) => { | ||
| this.writeResult(message); | ||
| }); | ||
| return super.run(); | ||
| } | ||
|
|
||
| shutdown() { | ||
| this._client.client.unsubscribe(); | ||
| } | ||
| } | ||
|
|
||
| class PatternSubscribeExecutor extends SubscribeExecutor { | ||
| constructor(client, commands) { | ||
| super(client, commands); | ||
| } | ||
|
|
||
| run() { | ||
| this._client.client.on("psubscribe", (pattern, count) => {}); | ||
|
|
||
| this._client.client.on("pmessage", (pattern, channel, message) => { | ||
| this.writeResult(message); | ||
| }); | ||
| return super.run(); | ||
| } | ||
| } | ||
|
|
||
| class MonitorExecutor extends Executor { | ||
| constructor(client, commands) { | ||
| super(client, commands); | ||
| } | ||
|
|
||
| run() { | ||
| this._client.client.on("monitor", (time, args, raw_reply) => { | ||
| this.writeResult(raw_reply); | ||
| }); | ||
| return super.run(); | ||
| } | ||
| } | ||
|
|
||
| module.exports = function (client, commands) { | ||
| const CMD = commands[0].toLowerCase(); | ||
| if (CMD === 'subscribe') { | ||
| return new SubscribeExecutor(client, commands); | ||
| } else if (CMD === 'psubscribe') { | ||
| return new PatternSubscribeExecutor(client, commands); | ||
| } else if (CMD === 'monitor') { | ||
| return new MonitorExecutor(client, commands); | ||
| } else { | ||
| return new Executor(client, commands); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| const RedisClient = require('../lib/redis').RedisClient; | ||
| const __PR__ = require('../lib/redis').__PR__; | ||
| const _log = global.console.log; | ||
| const colors = require('colors'); | ||
| const readline = require('readline'); | ||
| const createExecutor = require('../lib/executor') | ||
|
|
||
| let spy = {}; | ||
|
|
||
| let redisClient = {}; | ||
|
|
||
| beforeAll(() => { | ||
| redisClient = new RedisClient("127.0.0.1", 6379); | ||
| /** | ||
| * Call unref() on the underlying socket connection to the Redis server, allowing the program to exit once no more commands are pending. | ||
| * This is an experimental feature, as documented in the following site: | ||
| * https://github.com/NodeRedis/node_redis#clientunref | ||
| */ | ||
| redisClient.client.unref(); | ||
| // remove all listener to avoid async callback | ||
| redisClient.client.removeAllListeners(); | ||
| // mock `console.log` | ||
| spy.next = jest.spyOn(redisClient, 'next', 'set').mockImplementation(() => {}); | ||
| spy.exit = jest.spyOn(process, 'exit').mockImplementation(() => {}); | ||
| return redisClient.execute(['flushall']); | ||
| }); | ||
|
|
||
| afterAll(() => { | ||
| redisClient.client.quit(); | ||
| redisClient.rl.close(); | ||
| spy.next.mockRestore(); | ||
| spy.exit.mockRestore(); | ||
| }); | ||
|
|
||
| describe('subscribe executor', () => { | ||
| let pub = {}; | ||
| beforeAll(() => { | ||
| pub = new RedisClient("127.0.0.1", 6379); | ||
| pub.client.removeAllListeners(); | ||
| }); | ||
| it('subscribe to a channel', () => { | ||
| const sub = createExecutor(redisClient, ['subscribe', 'channel0']); | ||
| return sub.run().then(() => { | ||
| expect(spy.next).toBeCalledWith('channel0'); | ||
| }); | ||
| }) | ||
| }) |