Skip to content

Commit a7ebe9b

Browse files
habarthierry-huehabarthierry-hue
committed
Ticket #103 : Launch one or more EventMeshServer + add VPN + add client
1 parent 12a8fc9 commit a7ebe9b

86 files changed

Lines changed: 797 additions & 2813 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎FaasNet.EventMesh.sln‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11

22
Microsoft Visual Studio Solution File, Format Version 12.00
3-
# Visual Studio Version 16
4-
VisualStudioVersion = 16.0.31624.102
3+
# Visual Studio Version 17
4+
VisualStudioVersion = 17.1.32328.378
55
MinimumVisualStudioVersion = 10.0.40219.1
66
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "02. Core Layer", "02. Core Layer", "{6E495E0A-0DC8-4E42-8C58-3C48506D3D24}"
77
EndProject
@@ -39,6 +39,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "FaasNet.RaftConsensus.Core"
3939
EndProject
4040
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "FaasNet.RaftConsensus.Client", "src\RaftConsensus\FaasNet.RaftConsensus.Client\FaasNet.RaftConsensus.Client.csproj", "{9FD70B89-220D-47E7-958D-E10F493F9436}"
4141
EndProject
42+
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}"
43+
EndProject
4244
Global
4345
GlobalSection(SolutionConfigurationPlatforms) = preSolution
4446
Debug|Any CPU = Debug|Any CPU
@@ -97,6 +99,10 @@ Global
9799
{9FD70B89-220D-47E7-958D-E10F493F9436}.Debug|Any CPU.Build.0 = Debug|Any CPU
98100
{9FD70B89-220D-47E7-958D-E10F493F9436}.Release|Any CPU.ActiveCfg = Release|Any CPU
99101
{9FD70B89-220D-47E7-958D-E10F493F9436}.Release|Any CPU.Build.0 = Release|Any CPU
102+
{C212B250-18DD-4F63-B0B3-003FA45A8A9A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
103+
{C212B250-18DD-4F63-B0B3-003FA45A8A9A}.Debug|Any CPU.Build.0 = Debug|Any CPU
104+
{C212B250-18DD-4F63-B0B3-003FA45A8A9A}.Release|Any CPU.ActiveCfg = Release|Any CPU
105+
{C212B250-18DD-4F63-B0B3-003FA45A8A9A}.Release|Any CPU.Build.0 = Release|Any CPU
100106
EndGlobalSection
101107
GlobalSection(SolutionProperties) = preSolution
102108
HideSolutionNode = FALSE
@@ -115,6 +121,7 @@ Global
115121
{BE0F462B-1C13-4160-8A71-7FCDDE0C6BF3} = {6E495E0A-0DC8-4E42-8C58-3C48506D3D24}
116122
{69259CDF-1E32-4310-9155-BC095FFDBECD} = {7A13262A-D1E6-4210-BA95-03A0741CBFA4}
117123
{9FD70B89-220D-47E7-958D-E10F493F9436} = {7A13262A-D1E6-4210-BA95-03A0741CBFA4}
124+
{C212B250-18DD-4F63-B0B3-003FA45A8A9A} = {20A0BD99-A4F3-4FD5-A6FA-1935D7464DB8}
118125
EndGlobalSection
119126
GlobalSection(ExtensibilityGlobals) = postSolution
120127
SolutionGuid = {B9BD3B8C-B2C9-468F-BF54-66BFE9B565EC}

‎src/EventMesh/FaasNet.EventMesh.Client/EventMeshClient.cs‎

Lines changed: 151 additions & 81 deletions
Original file line numberDiff line numberDiff line change
@@ -1,137 +1,207 @@
11
using CloudNative.CloudEvents;
2+
using FaasNet.EventMesh.Client.Exceptions;
23
using FaasNet.EventMesh.Client.Messages;
4+
using FaasNet.RaftConsensus.Client;
5+
using FaasNet.RaftConsensus.Client.Extensions;
36
using System;
47
using System.Collections.Generic;
5-
using System.Diagnostics;
8+
using System.Linq;
9+
using System.Net;
10+
using System.Net.Sockets;
611
using System.Text.Json;
712
using System.Threading;
813
using System.Threading.Tasks;
914

1015
namespace FaasNet.EventMesh.Client
1116
{
12-
public class EventMeshClient : IDisposable
17+
public class EventMeshClient
1318
{
14-
private readonly string _clientId;
15-
private readonly string _password;
16-
private readonly string _vpn;
17-
private readonly RuntimeClient _runtimeClient;
18-
private readonly int _bufferCloudEvents;
19-
private HelloResponse _publishSession;
20-
private HelloResponse _subscribeSession;
21-
22-
public EventMeshClient(string clientId, string password, string vpn = Constants.DefaultVpn, string url = Constants.DefaultUrl, int port = Constants.DefaultPort, int bufferCloudEvents = 1)
19+
private readonly IPAddress _ipAddr;
20+
private readonly int _port;
21+
22+
public EventMeshClient(string url = Constants.DefaultUrl, int port = Constants.DefaultPort)
2323
{
24-
_clientId = clientId;
25-
_password = password;
26-
_vpn = vpn;
27-
_bufferCloudEvents = bufferCloudEvents;
28-
_runtimeClient = new RuntimeClient(url, port);
24+
_ipAddr = IPAddressHelper.ResolveIPAddress(url);
25+
_port = port;
2926
}
3027

31-
public async Task Connect(CancellationToken cancellationToken = default(CancellationToken))
28+
public async Task Ping(CancellationToken cancellationToken = default(CancellationToken))
3229
{
33-
await _runtimeClient.HeartBeat(cancellationToken);
30+
using (var udpClient = new UdpClient())
31+
{
32+
var writeCtx = new WriteBufferContext();
33+
var package = PackageRequestBuilder.HeartBeat();
34+
package.Serialize(writeCtx);
35+
var payload = writeCtx.Buffer.ToArray();
36+
await udpClient.SendAsync(payload, payload.Count(), new IPEndPoint(_ipAddr, _port)).WithCancellation(cancellationToken);
37+
var resultPayload = await udpClient.ReceiveAsync().WithCancellation(cancellationToken);
38+
var readCtx = new ReadBufferContext(resultPayload.Buffer);
39+
var packageResult = Package.Deserialize(readCtx);
40+
EnsureSuccessStatus(package, packageResult);
41+
}
3442
}
3543

3644
public async Task<IEnumerable<string>> GetAllVpns(CancellationToken cancellationToken = default(CancellationToken))
3745
{
38-
var vpnResponse = await _runtimeClient.GetAllVpns(cancellationToken);
39-
return vpnResponse.Vpns;
46+
using (var udpClient = new UdpClient())
47+
{
48+
var writeCtx = new WriteBufferContext();
49+
var package = PackageRequestBuilder.GetAllVpns();
50+
package.Serialize(writeCtx);
51+
var payload = writeCtx.Buffer.ToArray();
52+
await udpClient.SendAsync(payload, payload.Count(), new IPEndPoint(_ipAddr, _port)).WithCancellation(cancellationToken);
53+
var resultPayload = await udpClient.ReceiveAsync().WithCancellation(cancellationToken);
54+
var readCtx = new ReadBufferContext(resultPayload.Buffer);
55+
var packageResult = Package.Deserialize(readCtx);
56+
EnsureSuccessStatus(package, packageResult);
57+
var result = packageResult as GetAllVpnResponse;
58+
return result.Vpns;
59+
}
4060
}
4161

42-
public Task Publish(string topicName, object obj, CancellationToken cancellationToken = default(CancellationToken))
62+
public async Task<IEventMeshClientPubSession> CreatePubSession(string clientId, CancellationToken cancellationToken = default(CancellationToken))
4363
{
44-
var cloudEvt = new CloudEvent
45-
{
46-
Id = Guid.NewGuid().ToString(),
47-
Subject = topicName,
48-
Source = new Uri("http://localhost"),
49-
Type = topicName,
50-
DataContentType = "application/json",
51-
Data = JsonSerializer.Serialize(obj),
52-
Time = DateTimeOffset.UtcNow
53-
};
54-
return Publish(topicName, cloudEvt, cancellationToken);
64+
var pubSession = await CreateSession(clientId, cancellationToken);
65+
return new EventMeshClientPubSession(pubSession, _ipAddr, _port);
5566
}
5667

57-
public async Task Publish(string topicName, CloudEvent cloudEvent, TimeSpan? expirationTimeSpan = null, bool isSessionInfinite = false, CancellationToken cancellationToken = default(CancellationToken))
68+
public async Task<IEventMeshClientSubSession> CreateSubSession(string clientId, CancellationToken cancellationToken = default(CancellationToken))
5869
{
59-
if (_publishSession == null)
70+
var subSession = await CreateSession(clientId, cancellationToken);
71+
return new EventMeshClientSubSession(subSession, _ipAddr, _port);
72+
}
73+
74+
public async Task AddVpn(string vpn, CancellationToken cancellationToken = default(CancellationToken))
75+
{
76+
using (var udpClient = new UdpClient())
6077
{
61-
_publishSession = await CreateSession(_clientId, _password, UserAgentPurpose.PUB, expirationTimeSpan, isSessionInfinite, cancellationToken);
78+
var writeCtx = new WriteBufferContext();
79+
var package = PackageRequestBuilder.AddVPN(vpn);
80+
package.Serialize(writeCtx);
81+
var payload = writeCtx.Buffer.ToArray();
82+
await udpClient.SendAsync(payload, payload.Count(), new IPEndPoint(_ipAddr, _port)).WithCancellation(cancellationToken);
83+
var resultPayload = await udpClient.ReceiveAsync().WithCancellation(cancellationToken);
84+
var readCtx = new ReadBufferContext(resultPayload.Buffer);
85+
var packageResult = Package.Deserialize(readCtx);
86+
EnsureSuccessStatus(package, packageResult);
6287
}
63-
64-
await _runtimeClient.PublishMessage(_clientId, _publishSession.SessionId, topicName, cloudEvent);
6588
}
6689

67-
public async Task<SubscriptionResult> SubscribeMessages<TMessage>(string topicName, Action<TMessage> callback, TimeSpan? expirationTimeSpan = null, bool isSessionInfinite = false, CancellationToken cancellationToken = default(CancellationToken)) where TMessage : class
90+
public async Task AddClient(string vpn, string clientId, CancellationToken cancellationToken = default(CancellationToken))
6891
{
69-
return await Subscribe(topicName, (msg) =>
92+
using (var udpClient = new UdpClient())
7093
{
71-
foreach(var cloudEvt in msg.CloudEvents)
72-
{
73-
var deserialize = JsonSerializer.Deserialize(cloudEvt.Data.ToString(), typeof(TMessage)) as TMessage;
74-
callback(deserialize);
75-
}
76-
}, expirationTimeSpan, isSessionInfinite, cancellationToken);
94+
var writeCtx = new WriteBufferContext();
95+
var package = PackageRequestBuilder.AddClient(vpn, clientId);
96+
package.Serialize(writeCtx);
97+
var payload = writeCtx.Buffer.ToArray();
98+
await udpClient.SendAsync(payload, payload.Count(), new IPEndPoint(_ipAddr, _port)).WithCancellation(cancellationToken);
99+
var resultPayload = await udpClient.ReceiveAsync().WithCancellation(cancellationToken);
100+
var readCtx = new ReadBufferContext(resultPayload.Buffer);
101+
var packageResult = Package.Deserialize(readCtx);
102+
EnsureSuccessStatus(package, packageResult);
103+
}
77104
}
78105

79-
public async Task<SubscriptionResult> Subscribe(string topicName, Action<AsyncMessageToClient> callback, TimeSpan? expirationTimeSpan = null, bool isSessionInfinite = false, CancellationToken cancellationToken = default(CancellationToken))
106+
internal static void EnsureSuccessStatus(Package packageRequest, Package packageResponse)
80107
{
81-
if (_subscribeSession == null)
108+
if (packageResponse.Header.Status != HeaderStatus.SUCCESS)
82109
{
83-
_subscribeSession = await CreateSession(_clientId, _password, UserAgentPurpose.SUB, expirationTimeSpan, isSessionInfinite, cancellationToken);
110+
throw new RuntimeClientResponseException(packageResponse.Header.Status, packageResponse.Header.Error);
84111
}
85112

86-
return await _runtimeClient.Subscribe(_clientId, _subscribeSession.SessionId, new List<SubscriptionItem>
113+
if (packageRequest.Header.Seq != packageRequest.Header.Seq)
87114
{
88-
new SubscriptionItem
89-
{
90-
Topic = topicName
91-
}
92-
}, callback, cancellationToken: cancellationToken);
115+
throw new RuntimeClientResponseException(HeaderStatus.FAIL, Errors.INVALID_SEQ, "the seq in the request doesn't match the seq in the response");
116+
}
93117
}
94118

95-
public async Task Disconnect()
119+
private async Task<HelloResponse> CreateSession(string clientId, CancellationToken cancellationToken = default(CancellationToken))
96120
{
97-
if (_publishSession != null)
121+
var userAgent = new UserAgent { ClientId = clientId, Purpose = UserAgentPurpose.PUB };
122+
using (var udpClient = new UdpClient())
98123
{
99-
await _runtimeClient.Disconnect(_clientId, _publishSession.SessionId);
100-
}
101-
102-
if (_subscribeSession != null)
103-
{
104-
await _runtimeClient.Disconnect(_clientId, _subscribeSession.SessionId);
124+
var writeCtx = new WriteBufferContext();
125+
var package = PackageRequestBuilder.Hello(userAgent);
126+
package.Serialize(writeCtx);
127+
var payload = writeCtx.Buffer.ToArray();
128+
await udpClient.SendAsync(payload, payload.Count(), new IPEndPoint(_ipAddr, _port)).WithCancellation(cancellationToken);
129+
var resultPayload = await udpClient.ReceiveAsync().WithCancellation(cancellationToken);
130+
var readCtx = new ReadBufferContext(resultPayload.Buffer);
131+
var packageResult = Package.Deserialize(readCtx);
132+
EnsureSuccessStatus(package, packageResult);
133+
return packageResult as HelloResponse;
105134
}
106135
}
136+
}
107137

108-
public void Close()
138+
public interface IEventMeshClientPubSession
139+
{
140+
Task Publish(string topicName, object obj, CancellationToken cancellationToken = default(CancellationToken));
141+
Task Publish(string topicName, CloudEvent cloudEvent, CancellationToken cancellationToken = default(CancellationToken));
142+
}
143+
144+
public interface IEventMeshClientSubSession
145+
{
146+
147+
}
148+
149+
internal class EventMeshClientPubSession : IEventMeshClientPubSession
150+
{
151+
private readonly HelloResponse _session;
152+
private readonly IPAddress _ipAddr;
153+
private readonly int _port;
154+
155+
public EventMeshClientPubSession(HelloResponse session, IPAddress ipAddr, int port)
109156
{
110-
_runtimeClient.Close();
157+
_session = session;
158+
_ipAddr = ipAddr;
159+
_port = port;
111160
}
112161

113-
public void Dispose()
162+
public Task Publish(string topicName, object obj, CancellationToken cancellationToken = default(CancellationToken))
114163
{
115-
Disconnect().Wait();
116-
_runtimeClient.Close();
164+
var cloudEvt = new CloudEvent
165+
{
166+
Id = Guid.NewGuid().ToString(),
167+
Subject = topicName,
168+
Source = new Uri("http://localhost"),
169+
Type = topicName,
170+
DataContentType = "application/json",
171+
Data = JsonSerializer.Serialize(obj),
172+
Time = DateTimeOffset.UtcNow
173+
};
174+
return Publish(topicName, cloudEvt, cancellationToken);
117175
}
118176

119-
private Task<HelloResponse> CreateSession(string clientId, string password, UserAgentPurpose purpose, TimeSpan? expirationTimeSpan = null, bool isSessionInfinite = false, CancellationToken cancellationToken = default(CancellationToken))
177+
public async Task Publish(string topicName, CloudEvent cloudEvent, CancellationToken cancellationToken = default(CancellationToken))
120178
{
121-
var processId = Process.GetCurrentProcess().Id;
122-
return _runtimeClient.Hello(new UserAgent
179+
using (var udpClient = new UdpClient())
123180
{
124-
ClientId = clientId,
125-
Environment = "TST",
126-
Password = password,
127-
Pid = processId,
128-
Purpose = purpose,
129-
Version = "0",
130-
BufferCloudEvents = _bufferCloudEvents,
131-
Expiration = expirationTimeSpan,
132-
IsSessionInfinite = isSessionInfinite,
133-
Vpn = _vpn
134-
}, cancellationToken);
181+
var writeCtx = new WriteBufferContext();
182+
var package = PackageRequestBuilder.PublishMessage(_session.SessionId, topicName, cloudEvent);
183+
package.Serialize(writeCtx);
184+
var payload = writeCtx.Buffer.ToArray();
185+
await udpClient.SendAsync(payload, payload.Count(), new IPEndPoint(_ipAddr, _port));
186+
var resultPayload = await udpClient.ReceiveAsync();
187+
var readCtx = new ReadBufferContext(resultPayload.Buffer);
188+
var packageResult = Package.Deserialize(readCtx);
189+
EventMeshClient.EnsureSuccessStatus(package, packageResult);
190+
}
191+
}
192+
}
193+
194+
internal class EventMeshClientSubSession : IEventMeshClientSubSession
195+
{
196+
private readonly HelloResponse _session;
197+
private readonly IPAddress _ipAddr;
198+
private readonly int _port;
199+
200+
public EventMeshClientSubSession(HelloResponse session, IPAddress ipAddr, int port)
201+
{
202+
_session = session;
203+
_ipAddr = ipAddr;
204+
_port = port;
135205
}
136206
}
137207
}

‎src/EventMesh/FaasNet.EventMesh.Client/Extensions/AsyncExtensions.cs‎

Lines changed: 0 additions & 41 deletions
This file was deleted.

‎src/EventMesh/FaasNet.EventMesh.Client/FaasNet.EventMesh.Client.csproj‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,4 +6,7 @@
66
<PackageReference Include="CloudNative.CloudEvents" Version="2.2.0" />
77
<PackageReference Include="CloudNative.CloudEvents.SystemTextJson" Version="2.2.0" />
88
</ItemGroup>
9+
<ItemGroup>
10+
<ProjectReference Include="..\..\RaftConsensus\FaasNet.RaftConsensus.Client\FaasNet.RaftConsensus.Client.csproj" />
11+
</ItemGroup>
912
</Project>

‎src/EventMesh/FaasNet.EventMesh.Client/Messages/AddVpnRequest.cs‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
namespace FaasNet.EventMesh.Client.Messages
2+
{
3+
public class AddVpnRequest : Package
4+
{
5+
public string Vpn { get; set; }
6+
7+
public override void Serialize(WriteBufferContext context)
8+
{
9+
base.Serialize(context);
10+
context.WriteString(Vpn);
11+
}
12+
13+
public void Extract(ReadBufferContext context)
14+
{
15+
Vpn = context.NextString();
16+
}
17+
}
18+
}

0 commit comments

Comments
 (0)