Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 42 additions & 10 deletions Realtime/RealtimeChannel.cs
Original file line number Diff line number Diff line change
Expand Up @@ -541,21 +541,53 @@ public Push Push(string eventName, string? type = null, object? payload = null,
/// <param name="timeoutMs"></param>
public Task<bool> Send(ChannelEventName eventName, string? type, object payload, int timeoutMs = DefaultTimeout)
{
var tsc = new TaskCompletionSource<bool>();
var push = Push(Core.Helpers.GetMappedToAttr(eventName).Mapping, type, payload, timeoutMs);
// The server only sends a `phx_reply` for a `broadcast` push when the channel joined
// with `config.broadcast.ack = true`. Without that, no reply will ever arrive, so
// waiting for one would hang forever - resolve as soon as the push has been dispatched.
return eventName == ChannelEventName.Broadcast && BroadcastOptions?.BroadcastAck != true
? Task.FromResult(true)
: PushAwaiter.Await(push);
}

/// <summary>
/// Awaits a reply for a given <see cref="Channel.Push"/>, resolving with whether a known reply was
/// received, or faulting with a <see cref="RealtimeException"/> if the push times out first.
///
/// Kept as instance methods (rather than local lambdas referencing each other) so there's no
/// self/mutually-referencing closure over locals that get assigned after the closure is created.
/// </summary>
private sealed class PushAwaiter
{
private readonly Push push;
private readonly TaskCompletionSource<bool> taskCompletion = new();

private PushAwaiter(Push push)
{
this.push = push;
this.push.AddMessageReceivedHandler(HandleMessageReceived);
this.push.OnTimeout += HandleTimeout;
}

var ev = Core.Helpers.GetMappedToAttr(eventName).Mapping;
var push = Push(ev, type, payload, timeoutMs);
public static Task<bool> Await(Push push) => new PushAwaiter(push).taskCompletion.Task;

IRealtimePush<RealtimeChannel, SocketResponse>.MessageEventHandler? messageCallback = null;
private void HandleMessageReceived(IRealtimePush<RealtimeChannel, SocketResponse> sender, SocketResponse message)
{
Detach();
taskCompletion.TrySetResult(message.Event != EventType.Unknown);
}

messageCallback = (_, message) =>
private void HandleTimeout(object sender, EventArgs e)
{
tsc.SetResult(message.Event != EventType.Unknown);
push.RemoveMessageReceivedHandler(messageCallback!);
};
Detach();
taskCompletion.TrySetException(new RealtimeException("Push Timeout") { Reason = FailureHint.Reason.PushTimeout });
}

push.AddMessageReceivedHandler(messageCallback);
return tsc.Task;
private void Detach()
{
push.RemoveMessageReceivedHandler(HandleMessageReceived);
push.OnTimeout -= HandleTimeout;
}
}

/// <summary>
Expand Down
21 changes: 19 additions & 2 deletions RealtimeTests/ChannelBroadcastTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ public void CleanupTest()
_socketClient!.Disconnect();
}

[TestMethod("Channel: Can listen for broadcast")]
[TestMethod(DisplayName = "Channel: Can listen for broadcast")]
public async Task ClientCanListenForBroadcast()
{
var tsc = new TaskCompletionSource<bool>();
Expand Down Expand Up @@ -75,7 +75,24 @@ public async Task ClientCanListenForBroadcast()
await Task.WhenAll(new[] { tsc.Task, tsc2.Task });
}

[TestMethod("Channel: Payload returns a modeled response (if possible)")]
[TestMethod(DisplayName = "Channel: Send resolves when broadcast ack is not explicitly enabled")]
public async Task ChannelSendResolvesWithoutExplicitAck()
{
// Mirrors the most natural usage: `client.Channel(name)` -> `Subscribe()` -> `Send()`,
// with no call to `Register<T>(broadcastAck: true)`. See supabase-community/realtime-csharp#38.
var channel = _socketClient!.Channel("no-ack-broadcast");
await channel.Subscribe();

var sendTask = channel.Send(Supabase.Realtime.Constants.ChannelEventName.Broadcast, "test_event",
new BroadcastExample { UserId = Guid.NewGuid().ToString() });

var winner = await Task.WhenAny(sendTask, Task.Delay(TimeSpan.FromSeconds(5)));

Assert.AreSame(sendTask, winner, "channel.Send() did not complete within 5s (issue #38: awaited broadcast send hangs).");
Assert.IsTrue(await sendTask);
}

[TestMethod(DisplayName = "Channel: Payload returns a modeled response (if possible)")]
public async Task ChannelPayloadReturnsModel()
{
var tsc = new TaskCompletionSource<bool>();
Expand Down
Loading