Skip to content

Commit ad5f0aa

Browse files
habarthierry-huehabarthierry-hue
committed
Ticket #103 : WIP
1 parent 71fde58 commit ad5f0aa

22 files changed

Lines changed: 474 additions & 56 deletions

‎FaasNet.EventMesh.sln‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "FaasNet.RaftConsensus.Clien
4141
EndProject
4242
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "FaasNet.EventMesh.Runtime.Console", "src\EventMesh\FaasNet.EventMesh.Runtime.Console\FaasNet.EventMesh.Runtime.Console.csproj", "{C212B250-18DD-4F63-B0B3-003FA45A8A9A}"
4343
EndProject
44+
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "FaasNet.RaftConsensus.RocksDB", "src\RaftConsensus\FaasNet.RaftConsensus.RocksDB\FaasNet.RaftConsensus.RocksDB.csproj", "{7C4A643C-FEE8-484E-BE0B-8842FB986F20}"
45+
EndProject
4446
Global
4547
GlobalSection(SolutionConfigurationPlatforms) = preSolution
4648
Debug|Any CPU = Debug|Any CPU
@@ -103,6 +105,10 @@ Global
103105
{C212B250-18DD-4F63-B0B3-003FA45A8A9A}.Debug|Any CPU.Build.0 = Debug|Any CPU
104106
{C212B250-18DD-4F63-B0B3-003FA45A8A9A}.Release|Any CPU.ActiveCfg = Release|Any CPU
105107
{C212B250-18DD-4F63-B0B3-003FA45A8A9A}.Release|Any CPU.Build.0 = Release|Any CPU
108+
{7C4A643C-FEE8-484E-BE0B-8842FB986F20}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
109+
{7C4A643C-FEE8-484E-BE0B-8842FB986F20}.Debug|Any CPU.Build.0 = Debug|Any CPU
110+
{7C4A643C-FEE8-484E-BE0B-8842FB986F20}.Release|Any CPU.ActiveCfg = Release|Any CPU
111+
{7C4A643C-FEE8-484E-BE0B-8842FB986F20}.Release|Any CPU.Build.0 = Release|Any CPU
106112
EndGlobalSection
107113
GlobalSection(SolutionProperties) = preSolution
108114
HideSolutionNode = FALSE
@@ -122,6 +128,7 @@ Global
122128
{69259CDF-1E32-4310-9155-BC095FFDBECD} = {7A13262A-D1E6-4210-BA95-03A0741CBFA4}
123129
{9FD70B89-220D-47E7-958D-E10F493F9436} = {7A13262A-D1E6-4210-BA95-03A0741CBFA4}
124130
{C212B250-18DD-4F63-B0B3-003FA45A8A9A} = {20A0BD99-A4F3-4FD5-A6FA-1935D7464DB8}
131+
{7C4A643C-FEE8-484E-BE0B-8842FB986F20} = {7A13262A-D1E6-4210-BA95-03A0741CBFA4}
125132
EndGlobalSection
126133
GlobalSection(ExtensibilityGlobals) = postSolution
127134
SolutionGuid = {B9BD3B8C-B2C9-468F-BF54-66BFE9B565EC}

‎src/EventMesh/FaasNet.EventMesh.Runtime.Console/FaasNet.EventMesh.Runtime.Console.csproj‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
</ItemGroup>
1414

1515
<ItemGroup>
16+
<ProjectReference Include="..\..\RaftConsensus\FaasNet.RaftConsensus.RocksDB\FaasNet.RaftConsensus.RocksDB.csproj" />
1617
<ProjectReference Include="..\FaasNet.EventMesh.Runtime\FaasNet.EventMesh.Runtime.csproj" />
1718
</ItemGroup>
1819
</Project>

‎src/EventMesh/FaasNet.EventMesh.Runtime.Console/Program.cs‎

Lines changed: 24 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,9 @@
22
using FaasNet.EventMesh.Client;
33
using FaasNet.EventMesh.Client.Messages;
44
using FaasNet.RaftConsensus.Client;
5-
using FaasNet.RaftConsensus.Client.Messages.Consensus;
65
using FaasNet.RaftConsensus.Core;
76
using FaasNet.RaftConsensus.Core.Models;
7+
using FaasNet.RaftConsensus.Core.Stores;
88
using Microsoft.Extensions.DependencyInjection;
99
using System.Collections.Concurrent;
1010

@@ -73,6 +73,19 @@ async Task DisplayMenu(ICollection<INodeHost> nodes)
7373
Console.WriteLine("- Enter 'addclient' to add a client");
7474
Console.WriteLine("- Enter 'publishmsg' to publish a message");
7575
Console.WriteLine("- Enter 'submsg' to subscribe");
76+
// Nous avons plusieurs stratégies :
77+
// le message doit être publié sur un topic persisté sur plusieurs noeuds.
78+
// le client peut appeler de façon régulière un seul topic pour récupérer tous les messages (aucun filtre possible).
79+
// session : sub.
80+
// subscription : type = subscribe, index, topic.
81+
// append message => topic => read message.
82+
// le client peut utiliser une session persistée qui va permettre de récupérer dans son propre storage l'ensemble des messages.
83+
// session : sub.
84+
// subscription: type = persisted, topic filter
85+
// append message => topic => route to specific group id.
86+
// https://towardsdatascience.com/the-design-of-an-event-store-8c751c47db6f
87+
// Chaque node peut avoir un ou plusieurs peers.
88+
7689
// Pouvoir souscrire à un topic.
7790
// Pouvoir publier sur un topic.
7891
// Le seed kafka peut directement publier sur eventmesh.
@@ -82,7 +95,7 @@ async Task DisplayMenu(ICollection<INodeHost> nodes)
8295
{
8396
foreach (var node in nodes)
8497
{
85-
var entityTypes = await node.NodeStateStore.GetAllEntityTypes(CancellationToken.None);
98+
var entityTypes = await node.NodeStateStore.GetAllLastEntityTypes(CancellationToken.None);
8699
foreach (var entityType in entityTypes)
87100
{
88101
Console.WriteLine($"Port = {node.Port}, Type = {entityType.EntityType}, Version = {entityType.EntityVersion}, Value = {entityType.Value}");
@@ -158,9 +171,18 @@ INodeHost BuildNodeHost(int port, bool isSeed = false)
158171
{
159172
var serverBuilder = new ServiceCollection()
160173
.AddEventMeshServer(o => o.Port = port);
174+
// .UseRocksDB(o => { o.SubPath = $"node{port}"; });
161175
if (isSeed) serverBuilder.SetNodeStates(new ConcurrentBag<NodeState> { new ClusterNode { Port = 4000, Url = "localhost" }.ToNodeState() });
162176
var serviceProvider = serverBuilder.Services
163177
// .AddLogging(l => l.AddConsole())
164178
.BuildServiceProvider();
179+
/*
180+
if(isSeed)
181+
{
182+
var nodeStateStore = serviceProvider.GetRequiredService<INodeStateStore>();
183+
nodeStateStore.Add(new ClusterNode { Port = 4000, Url = "localhost" }.ToNodeState());
184+
}
185+
*/
186+
165187
return serviceProvider.GetRequiredService<INodeHost>();
166188
}

‎src/EventMesh/FaasNet.EventMesh.Runtime/EventMeshNode.cs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,8 @@ protected override async Task HandleUDPPackage(UdpReceiveResult udpResult, Cance
6161
if(result.Status.HasFlag(EventMeshPackageResultStatus.ADD_PEER))
6262
{
6363
await AddPeer(result.Termid);
64-
await StartPeer(result.Termid);
64+
var peerHost = await StartPeer(result.Termid);
65+
if (result.LogRecord != null) await peerHost.AppendEntry(result.LogRecord, TokenSource.Token);
6566
}
6667

6768
if(result.Status.HasFlag(EventMeshPackageResultStatus.SEND_RESULT))
@@ -74,6 +75,7 @@ protected override async Task HandleUDPPackage(UdpReceiveResult udpResult, Cance
7475
using (var udpClient = new UdpClient())
7576
{
7677
await udpClient.SendAsync(resultPayload, resultPayload.Count(), udpResult.RemoteEndPoint).WithCancellation(TokenSource.Token);
78+
udpClient.Close();
7779
}
7880
}
7981

‎src/EventMesh/FaasNet.EventMesh.Runtime/EventMeshPeer.cs‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
using FaasNet.EventMesh.Runtime.Stores;
2+
using FaasNet.RaftConsensus.Core;
3+
using FaasNet.RaftConsensus.Core.Models;
4+
using FaasNet.RaftConsensus.Core.Stores;
5+
using Microsoft.Extensions.Logging;
6+
using Microsoft.Extensions.Options;
7+
using System.Collections.Generic;
8+
using System.Linq;
9+
using System.Net.Sockets;
10+
using System.Threading;
11+
using System.Threading.Tasks;
12+
13+
namespace FaasNet.EventMesh.Runtime
14+
{
15+
public class EventMeshPeer : BasePeerHost
16+
{
17+
private readonly IMessageExchangeStore _messageExchangeStore;
18+
private readonly IQueueStore _queueStore;
19+
20+
public EventMeshPeer(IMessageExchangeStore messageExchangeStore, IQueueStore queueStore, ILogger<BasePeerHost> logger, IOptions<ConsensusPeerOptions> options, IClusterStore clusterStore, ILogStore logStore, IPeerInfoStore peerStore) : base(logger, options, clusterStore, logStore, peerStore)
21+
{
22+
_messageExchangeStore = messageExchangeStore;
23+
_queueStore = queueStore;
24+
}
25+
26+
protected override Task<bool> HandlePackage(UdpReceiveResult udpResult)
27+
{
28+
return Task.FromResult(true);
29+
}
30+
31+
protected override Task Init(CancellationToken token)
32+
{
33+
return Task.CompletedTask;
34+
}
35+
36+
protected override async Task AddEntry(LogRecord logRecord, CancellationToken cancellationToken)
37+
{
38+
if (State != PeerStates.LEADER) return;
39+
var queueNames = await GetQueueNames(cancellationToken);
40+
await AppendMessage(queueNames, logRecord, cancellationToken);
41+
}
42+
43+
private async Task<IEnumerable<string>> GetQueueNames(CancellationToken cancellationToken)
44+
{
45+
var messageExchanges = await _messageExchangeStore.GetAll(cancellationToken);
46+
var queueNames = new List<string>();
47+
foreach(var messageExchange in messageExchanges)
48+
{
49+
if(messageExchange.IsMatch(Info.TermId)) queueNames.AddRange(messageExchange.QueueNames);
50+
}
51+
52+
return queueNames.Distinct();
53+
}
54+
55+
private async Task AppendMessage(IEnumerable<string> queueNames, LogRecord logRecord, CancellationToken cancellationToken)
56+
{
57+
foreach(var queueName in queueNames)
58+
{
59+
await _queueStore.Add(queueName, logRecord.Value, cancellationToken);
60+
}
61+
}
62+
}
63+
}

‎src/EventMesh/FaasNet.EventMesh.Runtime/Handlers/EventMeshPackageResult.cs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
using FaasNet.EventMesh.Client.Messages;
2+
using FaasNet.RaftConsensus.Core.Models;
23
using System;
34

45
namespace FaasNet.EventMesh.Runtime.Handlers
@@ -8,17 +9,18 @@ public class EventMeshPackageResult
89
public Package Package { get; private set; }
910
public string Termid { get; private set; }
1011
public EventMeshPackageResultStatus Status { get; private set; }
12+
public LogRecord LogRecord { get; private set; }
1113

1214
public static EventMeshPackageResult SendResult(Package package)
1315
{
1416
return new EventMeshPackageResult { Package = package, Status = EventMeshPackageResultStatus.SEND_RESULT };
1517
}
1618

17-
public static EventMeshPackageResult AddPeer(string termid, Package package = null)
19+
public static EventMeshPackageResult AddPeer(string termid, Package package = null, LogRecord logRecord = null)
1820
{
1921
var status = EventMeshPackageResultStatus.ADD_PEER;
2022
if (package != null) status |= EventMeshPackageResultStatus.SEND_RESULT;
21-
return new EventMeshPackageResult { Termid = termid, Status = status, Package = package };
23+
return new EventMeshPackageResult { Termid = termid, Status = status, Package = package, LogRecord = logRecord };
2224
}
2325
}
2426

‎src/EventMesh/FaasNet.EventMesh.Runtime/Handlers/PublishMessageRequestHandler.cs‎

Lines changed: 18 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
using FaasNet.EventMesh.Client.Extensions;
22
using FaasNet.EventMesh.Client.Messages;
33
using FaasNet.EventMesh.Runtime.Exceptions;
4-
using FaasNet.EventMesh.Runtime.Models;
54
using FaasNet.EventMesh.Runtime.Stores;
65
using FaasNet.RaftConsensus.Client;
76
using FaasNet.RaftConsensus.Core;
@@ -36,10 +35,20 @@ public async Task<EventMeshPackageResult> Run(Package package, IEnumerable<IPeer
3635
{
3736
var publishMessageRequest = package as PublishMessageRequest;
3837
await CheckSession(publishMessageRequest, cancellationToken);
39-
var queueNames = await GetQueueNames(publishMessageRequest, cancellationToken);
40-
await BroadcastMessage(publishMessageRequest, queueNames, cancellationToken);
4138
var result = PackageResponseBuilder.PublishMessage(package.Header.Seq);
42-
return EventMeshPackageResult.SendResult(result);
39+
if (CheckPeerExists(peers, publishMessageRequest.Topic))
40+
{
41+
await SendMessage(publishMessageRequest, cancellationToken);
42+
return EventMeshPackageResult.SendResult(result);
43+
}
44+
45+
var base64Message = publishMessageRequest.CloudEvent.SerializeBase64();
46+
return EventMeshPackageResult.AddPeer(publishMessageRequest.Topic, result, new LogRecord
47+
{
48+
Index = 0,
49+
Value = base64Message,
50+
InsertionDateTime = DateTime.UtcNow
51+
});
4352
}
4453

4554
private async Task CheckSession(PublishMessageRequest message, CancellationToken cancellationToken)
@@ -57,36 +66,21 @@ private async Task CheckSession(PublishMessageRequest message, CancellationToken
5766
}
5867
}
5968

60-
private async Task<IEnumerable<string>> GetQueueNames(PublishMessageRequest message, CancellationToken cancellationToken)
69+
private bool CheckPeerExists(IEnumerable<IPeerHost> peers, string topicName)
6170
{
62-
IEnumerable<MessageExchange> messageExchanges;
63-
using(var activity = EventMeshMeter.RequestActivitySource.StartActivity("Get all message exchange"))
64-
{
65-
messageExchanges = await _messageExchangeStore.GetAll(cancellationToken);
66-
activity?.SetStatus(ActivityStatusCode.Ok);
67-
}
68-
69-
var queueNames = new List<string>();
70-
foreach(var messageExchange in messageExchanges)
71-
{
72-
if (messageExchange.IsMatch(message.Topic)) queueNames.AddRange(messageExchange.QueueNames);
73-
}
74-
75-
return queueNames.Distinct();
71+
var peer = peers.FirstOrDefault(p => p.Info.TermId == topicName);
72+
return peer != null;
7673
}
7774

78-
private async Task BroadcastMessage(PublishMessageRequest message, IEnumerable<string> queueNames, CancellationToken cancellationToken)
75+
private async Task SendMessage(PublishMessageRequest message, CancellationToken cancellationToken)
7976
{
8077
var rndClusterNode = await GetRandomClusterNode(cancellationToken);
8178
var base64Message = message.CloudEvent.SerializeBase64();
8279
using (var activity = EventMeshMeter.RequestActivitySource.StartActivity("Broadcast message"))
8380
{
8481
using (var consensusClient = new ConsensusClient(rndClusterNode.Url, rndClusterNode.Port))
8582
{
86-
foreach (var queueName in queueNames)
87-
{
88-
await consensusClient.AppendEntry(queueName, base64Message, cancellationToken);
89-
}
83+
await consensusClient.AppendEntry(message.Topic, base64Message, cancellationToken);
9084
}
9185
}
9286
}

‎src/EventMesh/FaasNet.EventMesh.Runtime/Handlers/ReadNextMessageHandler.cs‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77
using FaasNet.RaftConsensus.Core;
88
using System;
99
using System.Collections.Generic;
10-
using System.Linq;
1110
using System.Threading;
1211
using System.Threading.Tasks;
1312

@@ -16,10 +15,12 @@ namespace FaasNet.EventMesh.Runtime.Handlers
1615
public class ReadNextMessageHandler : IMessageHandler
1716
{
1817
private readonly IClientSessionStore _clientSessionStore;
18+
private readonly IQueueStore _queueStore;
1919

20-
public ReadNextMessageHandler(IClientSessionStore clientSessionStore)
20+
public ReadNextMessageHandler(IClientSessionStore clientSessionStore, IQueueStore queueStore)
2121
{
2222
_clientSessionStore = clientSessionStore;
23+
_queueStore = queueStore;
2324
}
2425

2526
public Commands Command => Commands.READ_NEXT_MESSAGE_REQUEST;
@@ -57,10 +58,9 @@ private async Task<CloudEvent> ReadNextMessage(ClientSession session, IEnumerabl
5758
{
5859
using (var activity = EventMeshMeter.RequestActivitySource.StartActivity("Read next message"))
5960
{
60-
var selectedPeer = peers.Single(p => p.Info.TermId == session.Queue);
61-
var lastLog = await selectedPeer.LogStore.Get(session.EvtOffset, cancellationToken);
62-
if (lastLog == null) return null;
63-
var cloudEvt = Convert.FromBase64String(lastLog.Value).DeserializeCloudEvent();
61+
var lastLog = await _queueStore.Get(session.Queue, session.EvtOffset, cancellationToken);
62+
if (string.IsNullOrWhiteSpace(lastLog)) return null;
63+
var cloudEvt = Convert.FromBase64String(lastLog).DeserializeCloudEvent();
6464
activity?.SetStatus(System.Diagnostics.ActivityStatusCode.Ok);
6565
return cloudEvt;
6666
}

‎src/EventMesh/FaasNet.EventMesh.Runtime/Handlers/SubscribeMessageHandler.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ public async Task<EventMeshPackageResult> Run(Package package, IEnumerable<IPeer
2626
var activeSession = await CheckSession(subscriptionRequest, cancellationToken);
2727
await AddMessageExchanges(activeSession.ClientSession.Queue, subscriptionRequest, cancellationToken);
2828
var result = PackageResponseBuilder.Subscription(package.Header.Seq, activeSession.ClientSession.Queue);
29-
return EventMeshPackageResult.AddPeer(activeSession.ClientSession.Queue, result);
29+
return EventMeshPackageResult.SendResult(result);
3030
}
3131

3232
private async Task<ActiveSessionResult> CheckSession(SubscriptionRequest request, CancellationToken cancellationToken)

‎src/EventMesh/FaasNet.EventMesh.Runtime/Models/QueueMessage.cs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
namespace FaasNet.EventMesh.Runtime.Models
2+
{
3+
public class QueueMessage
4+
{
5+
public string Value { get; set; }
6+
}
7+
}

0 commit comments

Comments
 (0)