From fd3a20dfe967f8bbe8a18870a7a960637641ada1 Mon Sep 17 00:00:00 2001 From: mnordine Date: Mon, 12 Sep 2016 09:10:25 -0300 Subject: [PATCH 1/6] Fix hsetnx command Hash id was not being sent to redis. Therefore, any time this command was used, a redis error would be returned. --- lib/redis_client/redis_client.dart | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/redis_client/redis_client.dart b/lib/redis_client/redis_client.dart index 6c1de83..5d0af13 100644 --- a/lib/redis_client/redis_client.dart +++ b/lib/redis_client/redis_client.dart @@ -1447,7 +1447,7 @@ class RedisClient { * If field already exists, this operation has no effect. */ Future hsetnx(String hashId, String key, Object value) => - connection.sendCommandWithVariadicValues(RedisCommand.HSETNX, [ key ], + connection.sendCommandWithVariadicValues(RedisCommand.HSETNX, [ hashId, key ], serializer.serializeToList(value)).receiveBool(); /** From ec3ec94675dd405cec21e6a7bb725866feba3e4e Mon Sep 17 00:00:00 2001 From: Mark Nordine Date: Mon, 12 Sep 2016 14:52:17 -0300 Subject: [PATCH 2/6] Add setnx redis command --- lib/redis_client/redis_client.dart | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/lib/redis_client/redis_client.dart b/lib/redis_client/redis_client.dart index 5d0af13..d8e506c 100644 --- a/lib/redis_client/redis_client.dart +++ b/lib/redis_client/redis_client.dart @@ -305,6 +305,14 @@ class RedisClient { connection.sendCommand(RedisCommand.PSETEX, [ key, expireInMs.toString(), value ]).receiveStatus("OK"); + /// If [key] does not exist, the key is added with the value of [value], + /// and true is returned. + /// + /// If [key] does exist, this operation has no effect and false is returned. + Future setnx(String key, Object value) => + connection.sendCommandWithVariadicValues(RedisCommand.SETNX, [ key ], + serializer.serializeToList(value)).receiveBool(); + /** * Remove the existing timeout on key. * From 16b407cacae453832542b347a8189573f535cfdf Mon Sep 17 00:00:00 2001 From: Mark Nordine Date: Mon, 4 May 2020 16:19:43 -0300 Subject: [PATCH 3/6] Add isConnected getter to client, set to false on errors --- lib/redis_client/redis_client.dart | 2 ++ lib/redis_client/redis_connection.dart | 12 ++++++++++++ 2 files changed, 14 insertions(+) diff --git a/lib/redis_client/redis_client.dart b/lib/redis_client/redis_client.dart index d8e506c..68c53b3 100644 --- a/lib/redis_client/redis_client.dart +++ b/lib/redis_client/redis_client.dart @@ -62,6 +62,8 @@ class RedisClient { Map get stats => connection.stats; Future close() => connection.close(); + + bool get isConnected => connection.isConnected; diff --git a/lib/redis_client/redis_connection.dart b/lib/redis_client/redis_connection.dart index 6bbdf72..b676e76 100644 --- a/lib/redis_client/redis_connection.dart +++ b/lib/redis_client/redis_connection.dart @@ -78,6 +78,8 @@ abstract class RedisConnection { /// Unubscribes from [List] channels Future unsubscribe(List channels); + + var isConnected = false; } @@ -169,6 +171,8 @@ class _RedisConnection extends RedisConnection { .transform(_streamTransformerHandler.createTransformer()) .listen(_onRedisReply, onError: _onStreamError, onDone: _onStreamDone); + isConnected = true; + if (password != null) return _authenticate(password); }) .then((_) { @@ -189,6 +193,7 @@ class _RedisConnection extends RedisConnection { /// Closes the connection. Future close() { logger.fine("Closing connection."); + isConnected = false; return this.connected.then((_) => _socket.close()); } @@ -209,6 +214,7 @@ class _RedisConnection extends RedisConnection { /// Gets called when the socket has an error. void _onSocketError(err) { logger.warning("Socket error $err."); + close(); throw new RedisClientException("Socket error $err."); } @@ -269,11 +275,13 @@ class _RedisConnection extends RedisConnection { void _onStreamError(err) { logger.warning("Stream error $err"); _socket.close(); + isConnected = false; throw new RedisClientException("Received stream error $err."); } /// Gets called when the stream closes. void _onStreamDone() { + isConnected = false; logger.fine("Stream finished."); } @@ -321,6 +329,10 @@ class _RedisConnection extends RedisConnection { * rawSend([ "GET".codeUnits, UTF8.encode("keyname") ]).receiveBulkString().then((String value) { }); */ Receiver rawSend(List> cmdWithArgs) { + + if (!isConnected) + throw new RedisClientException('redis socket not connected'); + var response = new Receiver(); From 5047663954cdcb3cc4a90252604ebf82d3b3eaa0 Mon Sep 17 00:00:00 2001 From: Mark Nordine Date: Wed, 20 May 2020 16:42:07 -0300 Subject: [PATCH 4/6] Improve subscription callback, since it should take a Receiver --- lib/redis_client/redis_client.dart | 4 +++- lib/redis_client/redis_connection.dart | 6 +++--- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/lib/redis_client/redis_client.dart b/lib/redis_client/redis_client.dart index 68c53b3..2444d29 100644 --- a/lib/redis_client/redis_client.dart +++ b/lib/redis_client/redis_client.dart @@ -2,6 +2,8 @@ part of redis_client; const exclusive = '('; +typedef SubscriptionCallback = Function(Receiver receiver); + /** * The [RedisClient] is a high level class to access your redis server. * @@ -1577,7 +1579,7 @@ class RedisClient { * Subscribes to [List ] channels * with [Function] onMessage handler */ - Future subscribe(List channels, Function onMessage) => connection.subscribe(channels, onMessage); + Future subscribe(List channels, SubscriptionCallback onMessage) => connection.subscribe(channels, onMessage); /** * Unubscribes from [List] channels diff --git a/lib/redis_client/redis_connection.dart b/lib/redis_client/redis_connection.dart index b676e76..255287d 100644 --- a/lib/redis_client/redis_connection.dart +++ b/lib/redis_client/redis_connection.dart @@ -73,7 +73,7 @@ abstract class RedisConnection { /// Subscribes to [List] channels with [Function] onMessage handler - Future subscribe(List channels, Function onMessage); + Future subscribe(List channels, SubscriptionCallback onMessage); /// Unubscribes from [List] channels @@ -218,9 +218,9 @@ class _RedisConnection extends RedisConnection { throw new RedisClientException("Socket error $err."); } - Function _subscriptionHandler = null; + SubscriptionCallback _subscriptionHandler = null; - Future subscribe(List channels, Function onMessage){ + Future subscribe(List channels, SubscriptionCallback onMessage){ Completer subscribeCompleter = new Completer(); List args = new List () From 6353c8df1b251305125d4232fdc8e768ffbc6f10 Mon Sep 17 00:00:00 2001 From: Mark Nordine Date: Thu, 3 Dec 2020 17:30:41 -0400 Subject: [PATCH 5/6] Add missing return type for publish --- lib/redis_client/redis_client.dart | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/redis_client/redis_client.dart b/lib/redis_client/redis_client.dart index 2444d29..74c45c4 100644 --- a/lib/redis_client/redis_client.dart +++ b/lib/redis_client/redis_client.dart @@ -1590,7 +1590,7 @@ class RedisClient { * Publishes [String] message to [String] channel * map when key does not exist. */ - Future publish(String channel, String message) => + Future publish(String channel, String message) => connection.sendCommand(RedisCommand.PUBLISH, [channel,message]) .receiveInteger(); From b3740ad52a1d036ad3e7bcffb0ef674f79d2aa6f Mon Sep 17 00:00:00 2001 From: Mark Nordine Date: Tue, 31 Aug 2021 22:00:10 -0300 Subject: [PATCH 6/6] Add duration and NX to set command --- lib/redis_client/redis_client.dart | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/lib/redis_client/redis_client.dart b/lib/redis_client/redis_client.dart index 74c45c4..a9fa855 100644 --- a/lib/redis_client/redis_client.dart +++ b/lib/redis_client/redis_client.dart @@ -288,7 +288,17 @@ class RedisClient { Future getset(String key, String value) => connection.sendCommand(RedisCommand.GETSET, [ key, value ]).receiveBulkString(); /// Sets the value of given key. - Future set(String key, String value) => connection.sendCommand(RedisCommand.SET, [ key, value ]).receiveStatus("OK"); + Future set(String key, String value, {bool ifNotExists = false, Duration expiration}) async { + final list = [key, value]; + if (expiration != null) list.addAll(['PX', (expiration.inMilliseconds).toString()]); + if (ifNotExists) list.add('NX'); + try { + final reply = await connection.sendCommand(RedisCommand.SET, list).receiveStatus('OK'); + return reply == 'OK'; + } on RedisClientException catch (e) { + return false; + } + } /** * Sets a value and the expiration in one step.