Commit 4b03e8d2 authored by Nick Craver's avatar Nick Craver

Cleanup: RedisSubscriber

parent d6681264
......@@ -6,7 +6,7 @@
namespace StackExchange.Redis
{
partial class ConnectionMultiplexer
public partial class ConnectionMultiplexer
{
private readonly Dictionary<RedisChannel, Subscription> subscriptions = new Dictionary<RedisChannel, Subscription>();
......@@ -33,8 +33,7 @@ internal Task AddSubscription(RedisChannel channel, Action<RedisChannel, RedisVa
{
lock (subscriptions)
{
Subscription sub;
if (subscriptions.TryGetValue(channel, out sub))
if (subscriptions.TryGetValue(channel, out Subscription sub))
{
sub.Add(handler);
}
......@@ -45,7 +44,6 @@ internal Task AddSubscription(RedisChannel channel, Action<RedisChannel, RedisVa
var task = sub.SubscribeToServer(this, channel, flags, asyncState, false);
if (task != null) return task;
}
}
}
return CompletedTask<bool>.Default(asyncState);
......@@ -57,8 +55,7 @@ internal ServerEndPoint GetSubscribedServer(RedisChannel channel)
{
lock (subscriptions)
{
Subscription sub;
if (subscriptions.TryGetValue(channel, out sub))
if (subscriptions.TryGetValue(channel, out Subscription sub))
{
return sub.GetOwner();
}
......@@ -72,8 +69,7 @@ internal void OnMessage(RedisChannel subscription, RedisChannel channel, RedisVa
ICompletable completable = null;
lock (subscriptions)
{
Subscription sub;
if (subscriptions.TryGetValue(subscription, out sub))
if (subscriptions.TryGetValue(subscription, out Subscription sub))
{
completable = sub.ForInvoke(channel, payload);
}
......@@ -101,17 +97,13 @@ internal Task RemoveSubscription(RedisChannel channel, Action<RedisChannel, Redi
{
lock (subscriptions)
{
Subscription sub;
if (subscriptions.TryGetValue(channel, out sub))
{
if (sub.Remove(handler))
if (subscriptions.TryGetValue(channel, out Subscription sub) && sub.Remove(handler))
{
subscriptions.Remove(channel);
var task = sub.UnsubscribeFromServer(channel, flags, asyncState, false);
if (task != null) return task;
}
}
}
return CompletedTask<bool>.Default(asyncState);
}
......@@ -133,11 +125,9 @@ internal bool SubscriberConnected(RedisChannel channel = default(RedisChannel))
if (server != null) return server.IsConnected;
server = SelectServer(-1, RedisCommand.SUBSCRIBE, CommandFlags.DemandMaster, default(RedisKey));
return server != null && server.IsConnected;
return server?.IsConnected == true;
}
internal long ValidateSubscriptions()
{
lock (subscriptions)
......@@ -156,14 +146,10 @@ private sealed class Subscription
private Action<RedisChannel, RedisValue> handler;
private ServerEndPoint owner;
public Subscription(Action<RedisChannel, RedisValue> value)
{
handler = value;
}
public void Add(Action<RedisChannel, RedisValue> value)
{
handler += value;
}
public Subscription(Action<RedisChannel, RedisValue> value) => handler = value;
public void Add(Action<RedisChannel, RedisValue> value) => handler += value;
public ICompletable ForInvoke(RedisChannel channel, RedisValue message)
{
var tmp = handler;
......@@ -182,6 +168,7 @@ public bool Remove(Action<RedisChannel, RedisValue> value)
return (handler -= value) == null;
}
}
public Task SubscribeToServer(ConnectionMultiplexer multiplexer, RedisChannel channel, CommandFlags flags, object asyncState, bool internalCall)
{
var cmd = channel.IsPatternBased ? RedisCommand.PSUBSCRIBE : RedisCommand.SUBSCRIBE;
......@@ -205,10 +192,8 @@ public Task UnsubscribeFromServer(RedisChannel channel, CommandFlags flags, obje
return oldOwner.QueueDirectAsync(msg, ResultProcessor.TrackSubscriptions, asyncState);
}
internal ServerEndPoint GetOwner()
{
return Interlocked.CompareExchange(ref owner, null, null);
}
internal ServerEndPoint GetOwner() => Interlocked.CompareExchange(ref owner, null, null);
internal void Resubscribe(RedisChannel channel, ServerEndPoint server)
{
if (server != null && Interlocked.CompareExchange(ref owner, server, server) == server)
......@@ -232,17 +217,12 @@ internal bool Validate(ConnectionMultiplexer multiplexer, RedisChannel channel)
}
oldOwner = null;
}
if (oldOwner == null)
{
if (SubscribeToServer(multiplexer, channel, CommandFlags.FireAndForget, null, true) != null)
if (oldOwner == null && SubscribeToServer(multiplexer, channel, CommandFlags.FireAndForget, null, true) != null)
{
changed = true;
}
}
return changed;
}
}
}
......@@ -309,12 +289,10 @@ public void Subscribe(RedisChannel channel, Action<RedisChannel, RedisValue> han
public Task SubscribeAsync(RedisChannel channel, Action<RedisChannel, RedisValue> handler, CommandFlags flags = CommandFlags.None)
{
if (channel.IsNullOrEmpty) throw new ArgumentNullException(nameof(channel));
return multiplexer.AddSubscription(channel, handler, flags, asyncState);
}
public EndPoint SubscribedEndpoint(RedisChannel channel)
{
var server = multiplexer.GetSubscribedServer(channel);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment