Skip to content

S
StackExchange.Redis
Commit c3c5234e authored by Marc Gravell's avatar Marc Gravell
Browse files
Options

expose the subscription channel on both the queue and the ChannelMessage;... 

expose the subscription channel on both the queue and the ChannelMessage; change the callbacks on ChannelMessageQueue to take ChannelMessage, to expose that
parent 86ee27f4
Hide whitespace changes
Inline Side-by-side
Showing with 46 additions and 37 deletions
StackExchange.Redis.Tests/Booksleeve/PubSub.cs
View file @ c3c5234e
......@@ -146,7 +146,7 @@ async Task RunLoop()
while (!subChannel.IsCompleted)
{
var work = await subChannel.ReadAsync();
int i = int.Parse(Encoding.UTF8.GetString(work.Value));
int i = int.Parse(Encoding.UTF8.GetString(work.Message));
lock (data)
{
data.Add(i);
......@@ -202,9 +202,9 @@ public async Task PubSubGetAllCorrectOrder_OnMessage_Sync()
var data = new List<int>(count);
var subChannel = await sub.SubscribeAsync(channel);
subChannel.OnMessage((key, val) =>
subChannel.OnMessage(msg =>
{
int i = int.Parse(Encoding.UTF8.GetString(val));
int i = int.Parse(Encoding.UTF8.GetString(msg.Message));
bool pulse = false;
lock (data)
{
......@@ -263,9 +263,9 @@ public async Task PubSubGetAllCorrectOrder_OnMessage_Async()
var data = new List<int>(count);
var subChannel = await sub.SubscribeAsync(channel);
subChannel.OnMessage((key, val) =>
subChannel.OnMessage(msg =>
{
int i = int.Parse(Encoding.UTF8.GetString(val));
int i = int.Parse(Encoding.UTF8.GetString(msg.Message));
bool pulse = false;
lock (data)
{
......
StackExchange.Redis/StackExchange/Redis/ChannelMessageQueue.cs
View file @ c3c5234e
......@@ -11,26 +11,33 @@ namespace StackExchange.Redis
/// </summary>
public readonly struct ChannelMessage
{
private readonly ChannelMessageQueue _queue; // this is *smaller* than storing a RedisChannel for the subsribed channel
/// <summary>
/// See Object.ToString
/// </summary>
public override string ToString() => ((string)Channel) + ":" + ((string)Value);
public override string ToString() => ((string)Channel) + ":" + ((string)Message);
/// <summary>
/// See Object.GetHashCode
/// </summary>
public override int GetHashCode() => Channel.GetHashCode() ^ Value.GetHashCode();
public override int GetHashCode() => Channel.GetHashCode() ^ Message.GetHashCode();
/// <summary>
/// See Object.Equals
/// </summary>
public override bool Equals(object obj) => obj is ChannelMessage cm
&& cm.Channel == Channel && cm.Value == Value;
internal ChannelMessage(RedisChannel channel, RedisValue value)
&& cm.Channel == Channel && cm.Message == Message;
internal ChannelMessage(ChannelMessageQueue queue, RedisChannel channel, RedisValue value)
{
_queue = queue;
Channel = channel;
Value = value;
Message = value;
}
/// <summary>
/// The channel that the subscription was created from
/// </summary>
public RedisChannel SubscriptionChannel => _queue.Channel;
/// <summary>
/// The channel that the message was broadcast to
/// </summary>
......@@ -38,7 +45,7 @@ internal ChannelMessage(RedisChannel channel, RedisValue value)
/// <summary>
/// The value that was broadcast
/// </summary>
public RedisValue Value { get; }
public RedisValue Message { get; }
}
......@@ -48,14 +55,17 @@ internal ChannelMessage(RedisChannel channel, RedisValue value)
/// <remarks>To create a ChannelMessageQueue, use ISubscriber.Subscribe[Async](RedisKey)</remarks>
public sealed class ChannelMessageQueue
{
private readonly Channel<ChannelMessage> _channel;
private readonly RedisChannel _redisChannel;
private readonly Channel<ChannelMessage> _queue;
/// <summary>
/// The Channel that was subscribed for this queue
/// </summary>
public RedisChannel Channel { get; }
private RedisSubscriber _parent;
/// <summary>
/// See Object.ToString
/// </summary>
public override string ToString() => (string)_redisChannel;
public override string ToString() => (string)Channel;
/// <summary>
/// Indicates if all messages that will be received have been drained from this channel
......@@ -64,10 +74,10 @@ public sealed class ChannelMessageQueue
internal ChannelMessageQueue(RedisChannel redisChannel, RedisSubscriber parent)
{
_redisChannel = redisChannel;
Channel = redisChannel;
_parent = parent;
_channel = Channel.CreateUnbounded<ChannelMessage>(s_ChannelOptions);
_channel.Reader.Completion.ContinueWith(
_queue = System.Threading.Channels.Channel.CreateUnbounded<ChannelMessage>(s_ChannelOptions);
_queue.Reader.Completion.ContinueWith(
(t, state) => ((ChannelMessageQueue)state).IsCompleted = true, this, TaskContinuationOptions.ExecuteSynchronously);
}
static readonly UnboundedChannelOptions s_ChannelOptions = new UnboundedChannelOptions
......@@ -76,19 +86,19 @@ internal ChannelMessageQueue(RedisChannel redisChannel, RedisSubscriber parent)
SingleReader = false,
AllowSynchronousContinuations = false,
};
internal void Subscribe(CommandFlags flags) => _parent.Subscribe(_redisChannel, HandleMessage, flags);
internal Task SubscribeAsync(CommandFlags flags) => _parent.SubscribeAsync(_redisChannel, HandleMessage, flags);
internal void Subscribe(CommandFlags flags) => _parent.Subscribe(Channel, HandleMessage, flags);
internal Task SubscribeAsync(CommandFlags flags) => _parent.SubscribeAsync(Channel, HandleMessage, flags);
private void HandleMessage(RedisChannel channel, RedisValue value)
{
var writer = _channel.Writer;
var writer = _queue.Writer;
if (channel.IsNull && value.IsNull) // see ForSyncShutdown
{
writer.TryComplete();
}
else
{
writer.TryWrite(new ChannelMessage(channel, value));
writer.TryWrite(new ChannelMessage(this, channel, value));
}
}
......@@ -97,12 +107,12 @@ private void HandleMessage(RedisChannel channel, RedisValue value)
/// Consume a message from the channel
/// </summary>
public ValueTask<ChannelMessage> ReadAsync(CancellationToken cancellationToken = default)
=> _channel.Reader.ReadAsync(cancellationToken);
=> _queue.Reader.ReadAsync(cancellationToken);
/// <summary>
/// Attempt to synchronously consume a message from the channel
/// </summary>
public bool TryRead(out ChannelMessage item) => _channel.Reader.TryRead(out item);
public bool TryRead(out ChannelMessage item) => _queue.Reader.TryRead(out item);
/// <summary>
/// Attempt to query the backlog length of the queue
......@@ -112,10 +122,10 @@ public bool TryGetCount(out int count)
// get this using the reflection
try
{
var prop = _channel.GetType().GetProperty("ItemsCountForDebugger", BindingFlags.Instance | BindingFlags.NonPublic);
var prop = _queue.GetType().GetProperty("ItemsCountForDebugger", BindingFlags.Instance | BindingFlags.NonPublic);
if (prop != null)
{
count = (int)prop.GetValue(_channel);
count = (int)prop.GetValue(_queue);
return true;
}
}
......@@ -134,16 +144,15 @@ private void AssertOnMessage(Delegate handler)
/// <summary>
/// Create a message loop that processes messages sequentially
/// </summary>
public void OnMessage(Action<RedisChannel, RedisValue> handler)
public void OnMessage(Action<ChannelMessage> handler)
{
AssertOnMessage(handler);
ThreadPool.QueueUserWorkItem(
state => ((ChannelMessageQueue)state).OnMessageSyncImpl(), this);
}
private async void OnMessageSyncImpl()
{
var handler = (Action<RedisChannel, RedisValue>)_onMessageHandler;
var handler = (Action<ChannelMessage>)_onMessageHandler;
while (!IsCompleted)
{
ChannelMessage next;
......@@ -155,7 +164,7 @@ private async void OnMessageSyncImpl()
break;
}
try { handler.Invoke(next.Channel, next.Value); }
try { handler(next); }
catch { } // matches MessageCompletable
}
}
......@@ -163,7 +172,7 @@ private async void OnMessageSyncImpl()
/// <summary>
/// Create a message loop that processes messages sequentially
/// </summary>
public void OnMessage(Func<RedisChannel, RedisValue, Task> handler)
public void OnMessage(Func<ChannelMessage, Task> handler)
{
AssertOnMessage(handler);
ThreadPool.QueueUserWorkItem(
......@@ -172,7 +181,7 @@ public void OnMessage(Func<RedisChannel, RedisValue, Task> handler)
private async void OnMessageAsyncImpl()
{
var handler = (Func<RedisChannel, RedisValue, Task>)_onMessageHandler;
var handler = (Func<ChannelMessage, Task>)_onMessageHandler;
while (!IsCompleted)
{
ChannelMessage next;
......@@ -186,8 +195,8 @@ private async void OnMessageAsyncImpl()
try
{
var task = handler.Invoke(next.Channel, next.Value);
if (task != null) await task.ConfigureAwait(false);
var task = handler(next);
if (task != null && task.Status != TaskStatus.RanToCompletion) await task.ConfigureAwait(false);
}
catch { } // matches MessageCompletable
}
......@@ -197,9 +206,9 @@ internal void UnsubscribeImpl(Exception error = null, CommandFlags flags = Comma
var parent = _parent;
if (parent != null)
{
parent.UnsubscribeAsync(_redisChannel, HandleMessage, flags);
parent.UnsubscribeAsync(Channel, HandleMessage, flags);
_parent = null;
_channel.Writer.TryComplete(error);
_queue.Writer.TryComplete(error);
}
}
internal async Task UnsubscribeAsyncImpl(Exception error = null, CommandFlags flags = CommandFlags.None)
......@@ -207,9 +216,9 @@ internal async Task UnsubscribeAsyncImpl(Exception error = null, CommandFlags fl
var parent = _parent;
if (parent != null)
{
await parent.UnsubscribeAsync(_redisChannel, HandleMessage, flags).ConfigureAwait(false);
await parent.UnsubscribeAsync(Channel, HandleMessage, flags).ConfigureAwait(false);
_parent = null;
_channel.Writer.TryComplete(error);
_queue.Writer.TryComplete(error);
}
}
......
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