Add OnMessageBytes to RedisSubscription + RedisPubSubServer

Author: mythz

src/ServiceStack.Redis/RedisPubSubServer.cs

@@ -30,6 +30,7 @@ public class RedisPubSubServer : IRedisPubSubServer
         /// Callback fired on each message received, handle with (channel, msg) => ... 
         /// </summary>
         public Action<string, string> OnMessage { get; set; }
+        public Action<string, byte[]> OnMessageBytes { get; set; }
 
         public Action<string> OnControlCommand { get; set; }
         public Action<string> OnUnSubscribe { get; set; }
@@ -55,8 +56,8 @@ public class RedisPubSubServer : IRedisPubSubServer
         private int autoRestart = YES;
         public bool AutoRestart
         {
-            get { return Interlocked.CompareExchange(ref autoRestart, 0, 0) == YES; }
-            set { Interlocked.CompareExchange(ref autoRestart, value ? YES : NO, autoRestart); }
+            get => Interlocked.CompareExchange(ref autoRestart, 0, 0) == YES;
+            set => Interlocked.CompareExchange(ref autoRestart, value ? YES : NO, autoRestart);
         }
 
         public DateTime CurrentServerTime => new DateTime(serverTimeAtStart.Ticks + startedAt.ElapsedTicks, DateTimeKind.Utc);
@@ -231,18 +232,35 @@ private void RunLoop()
                         {
                             subscription.OnUnSubscribe = HandleUnSubscribe;
 
+                            if (OnMessageBytes != null)
+                            {
+                                bool IsCtrlMessage(byte[] msg)
+                                {
+                                    if (msg.Length < 4)
+                                        return false;
+                                    return msg[0] == 'C' && msg[1] == 'T' && msg[0] == 'R' && msg[0] == 'L';
+                                }
+                                
+                                ((RedisSubscription)subscription).OnMessageBytes = (channel, msg) => {
+                                    if (IsCtrlMessage(msg))
+                                        return;
+
+                                    OnMessageBytes(channel, msg);
+                                };
+                            }
+
                             subscription.OnMessage = (channel, msg) =>
                             {
                                 if (string.IsNullOrEmpty(msg)) 
                                     return;
 
-                                var ctrlMsg = msg.SplitOnFirst(':');
-                                if (ctrlMsg[0] == ControlCommand.Control)
+                                var ctrlMsg = msg.LeftPart(':');
+                                if (ctrlMsg == ControlCommand.Control)
                                 {
                                     var op = Interlocked.CompareExchange(ref doOperation, Operation.NoOp, doOperation);
 
-                                    var msgType = ctrlMsg.Length > 1
-                                        ? ctrlMsg[1]
+                                    var msgType = msg.IndexOf(':') >= 0
+                                        ? msg.RightPart(':')
                                         : null;
 
                                     OnControlCommand?.Invoke(msgType ?? Operation.GetName(op));

src/ServiceStack.Redis/RedisSubscription.cs

@@ -30,6 +30,7 @@ public RedisSubscription(IRedisNativeClient redisClient)
 
         public Action<string> OnSubscribe { get; set; }
         public Action<string, string> OnMessage { get; set; }
+        public Action<string, byte[]> OnMessageBytes { get; set; }
         public Action<string> OnUnSubscribe { get; set; }
 
         public void SubscribeToChannels(params string[] channels)
@@ -91,17 +92,28 @@ private void ParseSubscriptionResults(byte[][] multiBytes)
                 }
                 else if (MessageWord.AreEqual(messageType))
                 {
-                    var message = multiBytes[i + MsgIndex].FromUtf8Bytes();
-
+                    var msgBytes = multiBytes[i + MsgIndex];
+                    if (this.OnMessageBytes != null)
+                    {
+                        this.OnMessageBytes(channel, msgBytes);
+                    }
+                    
+                    var message = msgBytes.FromUtf8Bytes();
                     if (this.OnMessage != null)
                     {
                         this.OnMessage(channel, message);
                     }
                 }
                 else if (PMessageWord.AreEqual(messageType))
                 {
-                    var message = multiBytes[i + MsgIndex + 1].FromUtf8Bytes();
                     channel = multiBytes[i + 2].FromUtf8Bytes();
+                    var msgBytes = multiBytes[i + MsgIndex + 1];
+                    if (this.OnMessageBytes != null)
+                    {
+                        this.OnMessageBytes(channel, msgBytes);
+                    }
+                    
+                    var message = msgBytes.FromUtf8Bytes();
                     if (this.OnMessage != null)
                     {
                         this.OnMessage(channel, message);