for pub/sub, treat length-one arrays comparably to simple values (#2118)

* fix #2117

for pub/sub, treak length-one arrays comparably to simple values

* update release notes

* actually check whether we're meant to be allowing arrays

Author: mgravell

docs/ReleaseNotes.md

@@ -29,6 +29,7 @@
 - Adds: Support for `OBJECT FREQ` with `.KeyFrequency()`/`.KeyFrequencyAsync()` ([#2105 by Avital-Fine](https://github.com/StackExchange/StackExchange.Redis/pull/2105))
 - Performance: Avoids allocations when computing cluster hash slots or testing key equality ([#2110 by Marc Gravell](https://github.com/StackExchange/StackExchange.Redis/pull/2110))
 - Adds: Support for `SORT_RO` with `.Sort()`/`.SortAsync()` ([#2111 by slorello89](https://github.com/StackExchange/StackExchange.Redis/pull/2111))
+- Adds: Support for pub/sub payloads that are unary arrays ([#2118 by Marc Gravell](https://github.com/StackExchange/StackExchange.Redis/pull/2118))
 
 ## 2.5.61
 

src/StackExchange.Redis/PhysicalConnection.cs

@@ -1498,10 +1498,10 @@ private void MatchResult(in RawResult result)
                     // invoke the handlers
                     var channel = items[1].AsRedisChannel(ChannelPrefix, RedisChannel.PatternMode.Literal);
                     Trace("MESSAGE: " + channel);
-                    if (!channel.IsNull)
+                    if (!channel.IsNull && TryGetPubSubPayload(items[2], out var payload))
                     {
                         _readStatus = ReadStatus.InvokePubSub;
-                        muxer.OnMessage(channel, channel, items[2].AsRedisValue());
+                        muxer.OnMessage(channel, channel, payload);
                     }
                     return; // AND STOP PROCESSING!
                 }
@@ -1511,11 +1511,11 @@ private void MatchResult(in RawResult result)
 
                     var channel = items[2].AsRedisChannel(ChannelPrefix, RedisChannel.PatternMode.Literal);
                     Trace("PMESSAGE: " + channel);
-                    if (!channel.IsNull)
+                    if (!channel.IsNull && TryGetPubSubPayload(items[3], out var payload))
                     {
                         var sub = items[1].AsRedisChannel(ChannelPrefix, RedisChannel.PatternMode.Pattern);
                         _readStatus = ReadStatus.InvokePubSub;
-                        muxer.OnMessage(sub, channel, items[3].AsRedisValue());
+                        muxer.OnMessage(sub, channel, payload);
                     }
                     return; // AND STOP PROCESSING!
                 }
@@ -1551,6 +1551,27 @@ private void MatchResult(in RawResult result)
             }
             _readStatus = ReadStatus.MatchResultComplete;
             _activeMessage = null;
+
+            static bool TryGetPubSubPayload(in RawResult value, out RedisValue parsed, bool allowArraySingleton = true)
+            {
+                if (value.IsNull)
+                {
+                    parsed = RedisValue.Null;
+                    return true;
+                }
+                switch (value.Type)
+                {
+                    case ResultType.Integer:
+                    case ResultType.SimpleString:
+                    case ResultType.BulkString:
+                        parsed = value.AsRedisValue();
+                        return true;
+                    case ResultType.MultiBulk when allowArraySingleton && value.ItemsCount == 1:
+                        return TryGetPubSubPayload(in value[0], out parsed, allowArraySingleton: false);
+                }
+                parsed = default;
+                return false;
+            }
         }
 
         private volatile Message? _activeMessage;