Fixed blocking stop() fct for tests

source/vibe/db/redis/redis.d

@@ -490,16 +490,14 @@ final class RedisSubscriber {
 			bool gotData;
 			StopWatch sw;
 			sw.start();
-			while (!gotData && !m_stop){
-				if (m_lockedConnection.conn.waitForData(5.seconds))
-					gotData = true;
-				if (sw.peek().seconds > timeout.total!"seconds") 
-					break;
-				if (m_stop) break;
+			while (!gotData){
+				if (m_lockedConnection.conn.waitForData(5.seconds))	gotData = true;
+				if (sw.peek().seconds > timeout.total!"seconds") { gotData = false;	break; }
+				if (m_stop){ gotData = false; break; }
 			}
 			sw.stop();
 
-			if (!gotData) break;
+			if (!gotData) { m_listening = false; m_lockedConnection.destroy(); return; }
 
 			if (m_capture !is null){
 				auto res = handler();
@@ -530,9 +528,6 @@ final class RedisSubscriber {
 			}
 		}
 
-		m_listening = false;
-		m_lockedConnection.destroy();
-
 	}
 	private Tuple!(long, string[]) handler(){
 		assert(m_lockedConnection !is null);

tests/redis/source/app.d

@@ -61,20 +61,22 @@ void runTest()
 	RedisSubscriber sub = new RedisSubscriber(redis);
 	import std.datetime;
 
+	assert(!sub.isListening);
 	sub.listen((string channel, string msg){
 		logInfo("LISTEN Recv Channel: %s, Message: %s", channel.to!string, msg.to!string);
 		logInfo("LISTEN Recv Time: %s", Clock.currTime().toString());
 	});
 
+	assert(sub.isListening);
 	sub.subscribe("SomeChannel");
 
 	logInfo("PUBLISH Sent: %s", Clock.currTime().toString());
 	redis.publish("SomeChannel", "Messageeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee");
 	sub.unsubscribe("SomeChannel");
-	auto stopped = sub.stop();
+	auto stopped = sub.bstop();
 	logInfo("LISTEN Stopped: %s", stopped.to!string);
+	assert(!sub.isListening);
 	redis.publish("SomeChannel", "Messageeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee");
-	sub.subscribe("SomeChannel");
 	sleep(1.seconds);
 	logInfo("Redis Test Succeeded.");
 }