Skip to content

Commit c66ed75

Browse files
tiasmtstephenlautier
tiasmt
and
stephenlautier
authored
feat(migration): update to orleans7 (#34)
Co-authored-by: Stephen Lautier <[email protected]>
1 parent fd5ac28 commit c66ed75

21 files changed

+218
-178
lines changed

‎.github/workflows/dotnet-publish.yml

+1-1
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ jobs:
2121
name: Package
2222
uses: sketch7/.github/.github/workflows/dotnet-package.yml@master
2323
with:
24-
dotnet-version: 6.0.x
24+
dotnet-version: 7.0.x
2525
publishable: ${{ contains(fromJSON('["develop", "master", "workflow", "feature/reusable-workflow"]'), github.ref_name) || endsWith(github.ref_name, '.x') }}
2626
secrets:
2727
nuget-auth-token: ${{ secrets.NUGET_KEY }}

‎Directory.Build.props

+3-3
Original file line numberDiff line numberDiff line change
@@ -31,17 +31,17 @@
3131
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
3232
<NoWarn>$(NoWarn);1701;1702;1705;1591</NoWarn>
3333
<AutoGenerateBindingRedirects>true</AutoGenerateBindingRedirects>
34-
<LangVersion>10.0</LangVersion>
34+
<LangVersion>11.0</LangVersion>
3535
</PropertyGroup>
3636

3737
<PropertyGroup>
38-
<TargetFramework>net6.0</TargetFramework>
38+
<TargetFramework>net7.0</TargetFramework>
3939
</PropertyGroup>
4040

4141
<!-- Shared Package Versions -->
4242
<PropertyGroup>
4343
<!-- vendors -->
44-
<OrleansVersion>3.6.0</OrleansVersion>
44+
<OrleansVersion>7.2.0</OrleansVersion>
4545
<OrleansStreamsUtilsVersion>11.0.0</OrleansStreamsUtilsVersion>
4646
</PropertyGroup>
4747
</Project>

‎package.json

+2-2
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"name": "@sketch7/signalr-orleans",
3-
"version": "4.0.4",
3+
"version": "5.0.0",
44
"versionSuffix": "",
55
"scripts": {
66
"pack": "bash ./tools/pack.sh",
@@ -9,4 +9,4 @@
99
"postpublish:dev": "rm -rf *.*nupkg",
1010
"test": "find test/**/*.csproj | xargs -i dotnet test {} --no-build -c Release --filter Category!=e2e"
1111
}
12-
}
12+
}

‎src/SignalR.Orleans.AspNet/OrleansHubLifetimeManager.cs

+19-25
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,20 @@
11
using Microsoft.AspNetCore.SignalR;
22
using Microsoft.AspNetCore.SignalR.Protocol;
33
using Microsoft.Extensions.Logging;
4-
using Orleans;
54
using Orleans.Concurrency;
65
using Orleans.Streams;
76
using SignalR.Orleans.Clients;
87
using SignalR.Orleans.Core;
98

109
namespace SignalR.Orleans;
1110

12-
public class OrleansHubLifetimeManager<THub> : HubLifetimeManager<THub>, IDisposable where THub : Hub
11+
public class OrleansHubLifetimeManager<THub> : HubLifetimeManager<THub>, IDisposable
12+
where THub : Hub
1313
{
1414
private Timer _timer;
1515
private readonly HubConnectionStore _connections = new HubConnectionStore();
1616
private readonly ILogger _logger;
17-
private readonly IClusterClientProvider _clusterClientProvider;
17+
private readonly IClusterClient _clusterClient;
1818
private readonly Guid _serverId;
1919
private IStreamProvider _streamProvider;
2020
private IAsyncStream<AllMessage> _allStream;
@@ -24,22 +24,22 @@ public class OrleansHubLifetimeManager<THub> : HubLifetimeManager<THub>, IDispos
2424

2525
public OrleansHubLifetimeManager(
2626
ILogger<OrleansHubLifetimeManager<THub>> logger,
27-
IClusterClientProvider clusterClientProvider
27+
IClusterClient clusterClient
2828
)
2929
{
30-
var hubType = typeof(THub).BaseType.GenericTypeArguments.FirstOrDefault() ?? typeof(THub);
30+
var hubType = typeof(THub).BaseType?.GenericTypeArguments.FirstOrDefault() ?? typeof(THub);
3131
_hubName = hubType.IsInterface && hubType.Name.StartsWith("I")
3232
? hubType.Name.Substring(1)
3333
: hubType.Name;
3434
_serverId = Guid.NewGuid(); // todo: include machine name
3535
_logger = logger;
36-
_clusterClientProvider = clusterClientProvider;
36+
_clusterClient = clusterClient;
3737
_ = EnsureStreamSetup();
3838
}
3939

4040
private Task HeartbeatCheck()
4141
{
42-
var client = _clusterClientProvider.GetClient().GetServerDirectoryGrain();
42+
var client = _clusterClient.GetServerDirectoryGrain();
4343
return client.Heartbeat(_serverId);
4444
}
4545

@@ -66,10 +66,10 @@ private async Task SetupStreams()
6666
{
6767
_logger.LogInformation("Initializing: Orleans HubLifetimeManager {hubName} (serverId: {serverId})...", _hubName, _serverId);
6868

69-
_streamProvider = _clusterClientProvider.GetClient().GetStreamProvider(Constants.STREAM_PROVIDER);
69+
_streamProvider = _clusterClient.GetStreamProvider(Constants.STREAM_PROVIDER);
7070
_serverStreamsReplicaContainer = new StreamReplicaContainer<ClientMessage>(_streamProvider, _serverId, Constants.SERVERS_STREAM, Constants.STREAM_SEND_REPLICAS);
7171

72-
_allStream = _streamProvider.GetStream<AllMessage>(Constants.ALL_STREAM_ID, Utils.BuildStreamHubName(_hubName));
72+
_allStream = _streamProvider.GetStream<AllMessage>(Utils.BuildStreamHubName(_hubName), Constants.ALL_STREAM_ID);
7373
_timer = new Timer(_ => Task.Run(HeartbeatCheck), null, TimeSpan.FromSeconds(0), TimeSpan.FromMinutes(Constants.HEARTBEAT_PULSE_IN_MINUTES));
7474

7575
var subscribeTasks = new List<Task>
@@ -120,15 +120,15 @@ public override async Task OnConnectedAsync(HubConnectionContext connection)
120120
{
121121
_connections.Add(connection);
122122

123-
var client = _clusterClientProvider.GetClient().GetClientGrain(_hubName, connection.ConnectionId);
123+
var client = _clusterClient.GetClientGrain(_hubName, connection.ConnectionId);
124124
await client.OnConnect(_serverId);
125125

126126
_logger.LogInformation("Connected {connectionId} on hub {hubName} with userId {userId} (serverId: {serverId})",
127127
connection.ConnectionId, _hubName, connection.UserIdentifier, _serverId);
128128

129129
if (connection.User.Identity.IsAuthenticated)
130130
{
131-
var user = _clusterClientProvider.GetClient().GetUserGrain(_hubName, connection.UserIdentifier);
131+
var user = _clusterClient.GetUserGrain(_hubName, connection.UserIdentifier);
132132
await user.Add(connection.ConnectionId);
133133
}
134134
}
@@ -147,7 +147,7 @@ public override async Task OnDisconnectedAsync(HubConnectionContext connection)
147147
{
148148
_logger.LogInformation("Disconnection {connectionId} on hub {hubName} with userId {userId} (serverId: {serverId})",
149149
connection.ConnectionId, _hubName, connection.UserIdentifier, _serverId);
150-
var client = _clusterClientProvider.GetClient().GetClientGrain(_hubName, connection.ConnectionId);
150+
var client = _clusterClient.GetClientGrain(_hubName, connection.ConnectionId);
151151
await client.OnDisconnect(ClientDisconnectReasons.HubDisconnect);
152152
}
153153
finally
@@ -196,7 +196,7 @@ public override Task SendGroupAsync(string groupName, string methodName, object[
196196
if (string.IsNullOrWhiteSpace(groupName)) throw new ArgumentNullException(nameof(groupName));
197197
if (string.IsNullOrWhiteSpace(methodName)) throw new ArgumentNullException(nameof(methodName));
198198

199-
var group = _clusterClientProvider.GetClient().GetGroupGrain(_hubName, groupName);
199+
var group = _clusterClient.GetGroupGrain(_hubName, groupName);
200200
return group.Send(methodName, args);
201201
}
202202

@@ -213,7 +213,7 @@ public override Task SendGroupExceptAsync(string groupName, string methodName, o
213213
if (string.IsNullOrWhiteSpace(groupName)) throw new ArgumentNullException(nameof(groupName));
214214
if (string.IsNullOrWhiteSpace(methodName)) throw new ArgumentNullException(nameof(methodName));
215215

216-
var group = _clusterClientProvider.GetClient().GetGroupGrain(_hubName, groupName);
216+
var group = _clusterClient.GetGroupGrain(_hubName, groupName);
217217
return group.SendExcept(methodName, args, excludedConnectionIds);
218218
}
219219

@@ -223,7 +223,7 @@ public override Task SendUserAsync(string userId, string methodName, object[] ar
223223
if (string.IsNullOrWhiteSpace(userId)) throw new ArgumentNullException(nameof(userId));
224224
if (string.IsNullOrWhiteSpace(methodName)) throw new ArgumentNullException(nameof(methodName));
225225

226-
var user = _clusterClientProvider.GetClient().GetUserGrain(_hubName, userId);
226+
var user = _clusterClient.GetUserGrain(_hubName, userId);
227227
return user.Send(methodName, args);
228228
}
229229

@@ -237,14 +237,14 @@ public override Task SendUsersAsync(IReadOnlyList<string> userIds, string method
237237
public override Task AddToGroupAsync(string connectionId, string groupName,
238238
CancellationToken cancellationToken = new CancellationToken())
239239
{
240-
var group = _clusterClientProvider.GetClient().GetGroupGrain(_hubName, groupName);
240+
var group = _clusterClient.GetGroupGrain(_hubName, groupName);
241241
return group.Add(connectionId);
242242
}
243243

244244
public override Task RemoveFromGroupAsync(string connectionId, string groupName,
245245
CancellationToken cancellationToken = new CancellationToken())
246246
{
247-
var group = _clusterClientProvider.GetClient().GetGroupGrain(_hubName, groupName);
247+
var group = _clusterClient.GetGroupGrain(_hubName, groupName);
248248
return group.Remove(connectionId);
249249
}
250250

@@ -257,7 +257,7 @@ private Task SendLocal(HubConnectionContext connection, HubInvocationMessage hub
257257

258258
private Task SendExternal(string connectionId, InvocationMessage hubMessage)
259259
{
260-
var client = _clusterClientProvider.GetClient().GetClientGrain(_hubName, connectionId);
260+
var client = _clusterClient.GetClientGrain(_hubName, connectionId);
261261
return client.Send(hubMessage.AsImmutable());
262262
}
263263

@@ -276,17 +276,11 @@ public void Dispose()
276276
toUnsubscribe.AddRange(subscriptions.Select(s => s.UnsubscribeAsync()));
277277
}
278278

279-
var serverDirectoryGrain = _clusterClientProvider.GetClient().GetServerDirectoryGrain();
279+
var serverDirectoryGrain = _clusterClient.GetServerDirectoryGrain();
280280
toUnsubscribe.Add(serverDirectoryGrain.Unregister(_serverId));
281281

282282
Task.WaitAll(toUnsubscribe.ToArray());
283283

284284
_timer?.Dispose();
285285
}
286286
}
287-
288-
public class AllMessage
289-
{
290-
public IReadOnlyList<string> ExcludedIds { get; set; }
291-
public InvocationMessage Payload { get; set; }
292-
}

‎src/SignalR.Orleans.AspNet/SignalR.Orleans.AspNet.csproj

+2-8
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
<PropertyGroup>
44
<PackageId>Sketch7.SignalR.Orleans.AspNet</PackageId>
5-
<TargetFramework>net6.0</TargetFramework>
5+
<TargetFramework>net7.0</TargetFramework>
66
</PropertyGroup>
77

88
<PropertyGroup>
@@ -13,13 +13,7 @@
1313
</PropertyGroup>
1414

1515
<ItemGroup>
16-
<PackageReference Include="Microsoft.Orleans.Core" Version="$(OrleansVersion)" />
17-
<PackageReference Include="Microsoft.Orleans.OrleansProviders" Version="$(OrleansVersion)" />
18-
<PackageReference Include="Microsoft.Orleans.Runtime.Abstractions" Version="$(OrleansVersion)" />
19-
<PackageReference Include="Microsoft.Orleans.CodeGenerator.MSBuild" Version="$(OrleansVersion)">
20-
<PrivateAssets>all</PrivateAssets>
21-
<IncludeAssets>runtime; build; native; contentfiles; analyzers</IncludeAssets>
22-
</PackageReference>
16+
<PackageReference Include="Microsoft.Orleans.Sdk" Version="$(OrleansVersion)" />
2317
</ItemGroup>
2418

2519
<!-- src libraries -->

‎src/SignalR.Orleans/Clients/ClientGrain.cs

+10-3
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
using Microsoft.AspNetCore.SignalR.Protocol;
2+
using Orleans.Runtime;
23
using SignalR.Orleans.Core;
34

45
namespace SignalR.Orleans.Clients;
@@ -31,10 +32,10 @@ public ClientGrain(ILogger<ClientGrain> logger)
3132
_logger = logger;
3233
}
3334

34-
public override async Task OnActivateAsync()
35+
public override async Task OnActivateAsync(CancellationToken cancellationToken)
3536
{
3637
_keyData = new ConnectionGrainKey(this.GetPrimaryKeyString());
37-
_streamProvider = GetStreamProvider(Constants.STREAM_PROVIDER);
38+
_streamProvider = this.GetStreamProvider(Constants.STREAM_PROVIDER);
3839

3940
if (State.ServerId == Guid.Empty)
4041
return;
@@ -76,6 +77,12 @@ public async Task Send(Immutable<InvocationMessage> message)
7677
}
7778
}
7879

80+
public Task SendOneWay(Immutable<InvocationMessage> message)
81+
{
82+
Send(message).Ignore();
83+
return Task.CompletedTask;
84+
}
85+
7986
public async Task OnConnect(Guid serverId)
8087
{
8188
State.ServerId = serverId;
@@ -111,6 +118,6 @@ private void SetupStreams()
111118
Constants.STREAM_SEND_REPLICAS,
112119
this.GetPrimaryKeyString()
113120
);
114-
_serverDisconnectedStream = _streamProvider.GetStream<Guid>(State.ServerId, Constants.SERVER_DISCONNECTED);
121+
_serverDisconnectedStream = _streamProvider.GetStream<Guid>(StreamId.Create(Constants.SERVER_DISCONNECTED, State.ServerId));
115122
}
116123
}

‎src/SignalR.Orleans/Clients/ClientMessage.cs

+4
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,13 @@
22

33
namespace SignalR.Orleans.Clients;
44

5+
[Immutable, GenerateSerializer]
56
public class ClientMessage
67
{
8+
[Id(0)]
79
public string HubName { get; set; }
10+
[Id(1)]
811
public string ConnectionId { get; set; }
12+
[Id(2), Immutable]
913
public InvocationMessage Payload { get; set; }
1014
}

‎src/SignalR.Orleans/Constants.cs

+2-2
Original file line numberDiff line numberDiff line change
@@ -11,8 +11,8 @@ public static class Constants
1111
public const string SERVERS_STREAM = "SERVERS_STREAM";
1212
public const string SERVER_DISCONNECTED = "SERVER_DISCONNECTED";
1313
public const string STREAM_PROVIDER = "ORLEANS_SIGNALR_STREAM_PROVIDER";
14-
public static readonly Guid CLIENT_DISCONNECT_STREAM_ID = Guid.Parse("bdcff7e7-3734-48ab-8599-17d915011b85");
15-
public static readonly Guid ALL_STREAM_ID = Guid.Parse("fbe53ecd-d896-4916-8281-5571d6733566");
14+
public static readonly string CLIENT_DISCONNECT_STREAM_ID = "CLIENT_DISCONNECT_STREAM";
15+
public static readonly string ALL_STREAM_ID = "ALL_STREAM";
1616

1717
internal const int STREAM_SEND_REPLICAS = 10; // todo: make configurable instead
1818
internal const double HEARTBEAT_PULSE_IN_MINUTES = 30;

‎src/SignalR.Orleans/Core/ConnectionGrain.cs

+16-13
Original file line numberDiff line numberDiff line change
@@ -25,11 +25,11 @@ internal ConnectionGrain(ILogger logger)
2525
_logger = logger;
2626
}
2727

28-
public override async Task OnActivateAsync()
28+
public override async Task OnActivateAsync(CancellationToken cancellationToken)
2929
{
3030
KeyData = new ConnectionGrainKey(this.GetPrimaryKeyString());
31-
_logger.Info("Activate {hubName} ({groupId})", KeyData.HubName, KeyData.Id);
32-
_streamProvider = GetStreamProvider(Constants.STREAM_PROVIDER);
31+
_logger.LogInformation("Activate {hubName} ({groupId})", KeyData.HubName, KeyData.Id);
32+
_streamProvider = this.GetStreamProvider(Constants.STREAM_PROVIDER);
3333

3434
_cleanupTimer = RegisterTimer(
3535
_ => CleanupStreams(),
@@ -49,9 +49,9 @@ public override async Task OnActivateAsync()
4949
}
5050
}
5151

52-
public override Task OnDeactivateAsync()
52+
public override Task OnDeactivateAsync(DeactivationReason reason, CancellationToken cancellationToken)
5353
{
54-
_logger.Info("Deactivate {hubName} ({groupId})", KeyData.HubName, KeyData.Id);
54+
_logger.LogInformation("Deactivate {hubName} ({groupId})", KeyData.HubName, KeyData.Id);
5555
_cleanupTimer?.Dispose();
5656
return CleanupStreams();
5757
}
@@ -60,7 +60,7 @@ public virtual async Task Add(string connectionId)
6060
{
6161
if (!State.Connections.Add(connectionId))
6262
return;
63-
_logger.Info("Added connection '{connectionId}' on {hubName} ({groupId}). {connectionsCount} connection(s)",
63+
_logger.LogInformation("Added connection '{connectionId}' on {hubName} ({groupId}). {connectionsCount} connection(s)",
6464
connectionId, KeyData.HubName, KeyData.Id, State.Connections.Count);
6565

6666
var clientDisconnectStream = GetClientDisconnectStream(connectionId);
@@ -71,7 +71,7 @@ public virtual async Task Add(string connectionId)
7171
public virtual async Task Remove(string connectionId)
7272
{
7373
var shouldWriteState = State.Connections.Remove(connectionId);
74-
_logger.Info("Removing connection '{connectionId}' on {hubName} ({groupId}). Remaining {connectionsCount} connection(s), was found: {isConnectionFound}",
74+
_logger.LogInformation("Removing connection '{connectionId}' on {hubName} ({groupId}). Remaining {connectionsCount} connection(s), was found: {isConnectionFound}",
7575
connectionId, KeyData.HubName, KeyData.Id, State.Connections.Count, shouldWriteState);
7676
_connectionStreamToUnsubscribe.Add(connectionId);
7777

@@ -88,6 +88,12 @@ public virtual async Task Remove(string connectionId)
8888
public virtual Task Send(Immutable<InvocationMessage> message)
8989
=> SendAll(message, State.Connections);
9090

91+
public Task SendOneWay(Immutable<InvocationMessage> message)
92+
{
93+
Send(message).Ignore();
94+
return Task.CompletedTask;
95+
}
96+
9197
public Task SendExcept(string methodName, object[] args, IReadOnlyList<string> excludedConnectionIds)
9298
{
9399
var message = new Immutable<InvocationMessage>(new InvocationMessage(methodName, args));
@@ -99,14 +105,11 @@ public Task<int> Count()
99105

100106
protected Task SendAll(Immutable<InvocationMessage> message, IReadOnlyCollection<string> connections)
101107
{
102-
_logger.Debug("Sending message to {hubName}.{targetMethod} on group {groupId} to {connectionsCount} connection(s)",
108+
_logger.LogDebug("Sending message to {hubName}.{targetMethod} on group {groupId} to {connectionsCount} connection(s)",
103109
KeyData.HubName, message.Value.Target, KeyData.Id, connections.Count);
104110

105-
foreach (var connection in connections)
106-
{
107-
GrainFactory.GetClientGrain(KeyData.HubName, connection)
108-
.InvokeOneWay(x => x.Send(message));
109-
}
111+
foreach (var connection in connections)
112+
GrainFactory.GetClientGrain(KeyData.HubName, connection).SendOneWay(message);
110113

111114
return Task.CompletedTask;
112115
}

‎src/SignalR.Orleans/Core/GrainExtensions.cs

+4-1
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,10 @@ public static Task Send(this IHubMessageInvoker grain, string methodName, params
2828
/// <param name="methodName">Target method name to invoke.</param>
2929
/// <param name="args">Arguments to pass to the target method.</param>
3030
public static void SendOneWay(this IHubMessageInvoker grain, string methodName, params object[] args)
31-
=> grain.InvokeOneWay(g => g.Send(methodName, args));
31+
{
32+
var invocationMessage = new InvocationMessage(methodName, args).AsImmutable();
33+
grain.SendOneWay(invocationMessage);
34+
}
3235
}
3336

3437
public static class GrainFactoryExtensions

0 commit comments

Comments
 (0)