diff --git a/lib/redis_client/redis_client.dart b/lib/redis_client/redis_client.dart index 6c1de83..a9fa855 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. * @@ -62,6 +64,8 @@ class RedisClient { Map get stats => connection.stats; Future close() => connection.close(); + + bool get isConnected => connection.isConnected; @@ -284,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. @@ -305,6 +319,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. * @@ -1447,7 +1469,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(); /** @@ -1567,7 +1589,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 @@ -1578,7 +1600,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(); diff --git a/lib/redis_client/redis_connection.dart b/lib/redis_client/redis_connection.dart index 6bbdf72..255287d 100644 --- a/lib/redis_client/redis_connection.dart +++ b/lib/redis_client/redis_connection.dart @@ -73,11 +73,13 @@ 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 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,12 +214,13 @@ 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."); } - 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 () @@ -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();