Add ChannelsMatching support to RedisPubServer for subscribing to channel patterns

mythz · mythz · commit d0bd39b79239 · 2015-07-13T15:06:08.000+08:00
Add IsSentinelSubscription to avoid calling unknown operations on
Sentinel hosts like TIME
diff --git a/src/ServiceStack.Redis/RedisPubSubServer.cs b/src/ServiceStack.Redis/RedisPubSubServer.cs
@@ -3,7 +3,6 @@
 using System.Text;
 using System.Threading;
 using ServiceStack.Logging;
-using ServiceStack.Text;
 
 namespace ServiceStack.Redis
 {
@@ -29,6 +28,7 @@ public class RedisPubSubServer : IRedisPubSubServer
         public Action<string> OnUnSubscribe { get; set; }
         public Action<Exception> OnError { get; set; }
         public Action<IRedisPubSubServer> OnFailover { get; set; }
+        public bool IsSentinelSubscription { get; set; }
 
         readonly Random rand = new Random(Environment.TickCount);
 
@@ -62,8 +62,10 @@ public long BgThreadCount
             get { return Interlocked.CompareExchange(ref bgThreadCount, 0, 0); }
         }
 
+        public const string AllChannelsWildCard = "*";
         public IRedisClientsManager ClientsManager { get; set; }
         public string[] Channels { get; set; }
+        public string[] ChannelsMatching { get; set; }
         public TimeSpan? WaitBeforeNextRestart { get; set; }
 
         public RedisPubSubServer(IRedisClientsManager clientsManager, params string[] channels)
@@ -141,7 +143,9 @@ private void Init()
                 using (var redis = ClientsManager.GetReadOnlyClient())
                 {
                     startedAt = Stopwatch.StartNew();
-                    serverTimeAtStart = redis.GetServerTime();
+                    serverTimeAtStart = IsSentinelSubscription
+                        ? DateTime.UtcNow
+                        : redis.GetServerTime();
                 }
             }
             catch (Exception ex)
@@ -281,7 +285,12 @@ private void RunLoop()
                                 }
                             };
 
-                            subscription.SubscribeToChannels(Channels); //blocks thread
+                            //blocks thread
+                            if (ChannelsMatching != null)
+                                subscription.SubscribeToChannelsMatching(ChannelsMatching);
+                            else
+                                subscription.SubscribeToChannels(Channels);             
+
                             masterClient = null;
                         }
                     }

Original file line number	Diff line number	Diff line change
`@@ -3,7 +3,6 @@`
`3`	`3`	`using System.Text;`
`4`	`4`	`using System.Threading;`
`5`	`5`	`using ServiceStack.Logging;`
`6`		`-using ServiceStack.Text;`
`7`	`6`
`8`	`7`	`namespace ServiceStack.Redis`
`9`	`8`	`{`
`@@ -29,6 +28,7 @@ public class RedisPubSubServer : IRedisPubSubServer`
`29`	`28`	`public Action<string> OnUnSubscribe { get; set; }`
`30`	`29`	`public Action<Exception> OnError { get; set; }`
`31`	`30`	`public Action<IRedisPubSubServer> OnFailover { get; set; }`
	`31`	`+ public bool IsSentinelSubscription { get; set; }`
`32`	`32`
`33`	`33`	`readonly Random rand = new Random(Environment.TickCount);`
`34`	`34`
`@@ -62,8 +62,10 @@ public long BgThreadCount`
`62`	`62`	`get { return Interlocked.CompareExchange(ref bgThreadCount, 0, 0); }`
`63`	`63`	`}`
`64`	`64`
	`65`	`+ public const string AllChannelsWildCard = "*";`
`65`	`66`	`public IRedisClientsManager ClientsManager { get; set; }`
`66`	`67`	`public string[] Channels { get; set; }`
	`68`	`+ public string[] ChannelsMatching { get; set; }`
`67`	`69`	`public TimeSpan? WaitBeforeNextRestart { get; set; }`
`68`	`70`
`69`	`71`	`public RedisPubSubServer(IRedisClientsManager clientsManager, params string[] channels)`
`@@ -141,7 +143,9 @@ private void Init()`
`141`	`143`	`using (var redis = ClientsManager.GetReadOnlyClient())`
`142`	`144`	`{`
`143`	`145`	`startedAt = Stopwatch.StartNew();`
`144`		`- serverTimeAtStart = redis.GetServerTime();`
	`146`	`+ serverTimeAtStart = IsSentinelSubscription`
	`147`	`+ ? DateTime.UtcNow`
	`148`	`+ : redis.GetServerTime();`
`145`	`149`	`}`
`146`	`150`	`}`
`147`	`151`	`catch (Exception ex)`
`@@ -281,7 +285,12 @@ private void RunLoop()`
`281`	`285`	`}`
`282`	`286`	`};`
`283`	`287`
`284`		`- subscription.SubscribeToChannels(Channels); //blocks thread`
	`288`	`+ //blocks thread`
	`289`	`+ if (ChannelsMatching != null)`
	`290`	`+ subscription.SubscribeToChannelsMatching(ChannelsMatching);`
	`291`	`+ else`
	`292`	`+ subscription.SubscribeToChannels(Channels);`
	`293`	`+`
`285`	`294`	`masterClient = null;`
`286`	`295`	`}`
`287`	`296`	`}`