Skip to content

Commit 8d84de1

Browse files
habarthierry-huehabarthierry-hue
committed
Ticket ## : Fix the UT
1 parent f9d2466 commit 8d84de1

File tree

12 files changed

+116
-81
lines changed

12 files changed

+116
-81
lines changed

‎src/EventMesh.Runtime.AMQP/AMQPConsumer.cs‎

Lines changed: 30 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -7,19 +7,19 @@
77
using RabbitMQ.Client.Events;
88
using System;
99
using System.Collections.Generic;
10+
using System.Linq;
1011
using System.Threading;
1112
using System.Threading.Tasks;
1213

1314
namespace EventMesh.Runtime.AMQP
1415
{
1516
public class AMQPConsumer : IMessageConsumer
1617
{
17-
private readonly List<string> _subscribedTopics = new List<string>();
18+
private readonly List<AMQPSubscriptionRecord> _subscriptions = new List<AMQPSubscriptionRecord>();
1819
private readonly IBrokerConfigurationStore _brokerConfigurationStore;
1920
private readonly AMQPOptions _opts;
2021
private readonly IClientStore _clientStore;
2122
private readonly RuntimeOptions _runtimeOpts;
22-
private static object _obj = new object();
2323
private IConnection _connection;
2424

2525
public event EventHandler<CloudEventArgs> CloudEventReceived;
@@ -38,6 +38,14 @@ public AMQPConsumer(
3838

3939
#region Actions
4040

41+
public string BrokerName
42+
{
43+
get
44+
{
45+
return _opts.BrokerName;
46+
}
47+
}
48+
4149
public Task Start(CancellationToken cancellationToken)
4250
{
4351
var options = GetOptions();
@@ -75,7 +83,7 @@ public Task Subscribe(string topicName, Client client, string sessionId, Cancell
7583
Task.Run(() =>
7684
{
7785
Thread.Sleep(_runtimeOpts.WaitLocalSubscriptionIntervalMS);
78-
ListenTopic(options, topicName, topic);
86+
ListenTopic(options, topicName, topic, client.ClientId, activeSession.Id);
7987
});
8088
activeSession.SubscribeTopic(topicName, options.BrokerName);
8189
return Task.CompletedTask;
@@ -90,7 +98,9 @@ public Task Unsubscribe(string topicName, Client client, string sessionId, Cance
9098
return Task.CompletedTask;
9199
}
92100

93-
activeSession.UnsubscribeTopic(topicName, options.BrokerName);
101+
var subscription = _subscriptions.First(s => s.ClientSessionId == sessionId && s.ClientId == client.ClientId && s.TopicName == topicName);
102+
subscription.Channel.BasicCancel(subscription.ConsumerTag);
103+
_subscriptions.Remove(subscription);
94104
return Task.CompletedTask;
95105
}
96106

@@ -101,11 +111,11 @@ public void Dispose()
101111

102112
#endregion
103113

104-
private IModel ListenTopic(AMQPOptions options, string topicName, ClientTopic topic)
114+
private void ListenTopic(AMQPOptions options, string topicName, ClientTopic topic, string clientId, string clientSessionId)
105115
{
106-
if(_subscribedTopics.Contains(topicName))
116+
if (_subscriptions.Any(s => s.BrokerName == options.BrokerName && s.TopicName == topicName && s.ClientSessionId == clientSessionId && s.ClientId == clientId))
107117
{
108-
return null;
118+
return;
109119
}
110120

111121
var channel = _connection.CreateModel();
@@ -120,28 +130,26 @@ private IModel ListenTopic(AMQPOptions options, string topicName, ClientTopic to
120130
});
121131
channel.QueueBind(queue, options.TopicName, topicName);
122132
var consumer = new EventingBasicConsumer(channel);
123-
consumer.Received += (sender, e) => ReceiveMessage(sender, topicName, options.Source, options.BrokerName, e);
133+
consumer.Received += (sender, e) => ReceiveMessage(sender, clientId, clientSessionId, topicName, options.Source, options.BrokerName, e);
124134
// TODO : Update BasicQos.
125135
channel.BasicQos(0, 100, false);
126-
channel.BasicConsume(queue, false, string.Empty, new Dictionary<string, object> { { "x-stream-offset", topic.Offset } }, consumer);
127-
_subscribedTopics.Add(topicName);
128-
return channel;
136+
var tag = channel.BasicConsume(queue, false, string.Empty, new Dictionary<string, object> { { "x-stream-offset", topic.Offset } }, consumer);
137+
_subscriptions.Add(new AMQPSubscriptionRecord(topicName, options.BrokerName, clientId, clientSessionId, channel, tag));
129138
}
130139

131-
private void ReceiveMessage(object sender, string topicName, string source, string brokerName, BasicDeliverEventArgs e)
140+
private void ReceiveMessage(object sender, string clientId, string clientSessionId, string topicName, string source, string brokerName, BasicDeliverEventArgs e)
132141
{
133-
lock(_obj)
142+
var jsonEventFormatter = new JsonEventFormatter();
143+
var model = (sender as EventingBasicConsumer).Model;
144+
var cloudEvent = e.ToCloudEvent(jsonEventFormatter, source, topicName);
145+
var client = _clientStore.GetByActiveSession(clientId, clientSessionId);
146+
if (client == null)
134147
{
135-
var jsonEventFormatter = new JsonEventFormatter();
136-
var model = (sender as EventingBasicConsumer).Model;
137-
var cloudEvent = e.ToCloudEvent(jsonEventFormatter, source, topicName);
138-
var activeClients = _clientStore.GetAllBySubscribedTopics(brokerName, topicName);
139-
foreach (var client in activeClients)
140-
{
141-
var clientSession = client.GetActiveSessionByTopic(brokerName, topicName);
142-
CloudEventReceived(this, new CloudEventArgs(topicName, brokerName, cloudEvent, client.ClientId, clientSession));
143-
}
148+
return;
144149
}
150+
151+
var clientSession = client.GetActiveSessionByTopic(brokerName, topicName);
152+
CloudEventReceived(this, new CloudEventArgs(topicName, brokerName, cloudEvent, client.ClientId, clientSession));
145153
}
146154

147155
private AMQPOptions GetOptions()

‎src/EventMesh.Runtime.AMQP/AMQPSubscriptionRecord.cs‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
using RabbitMQ.Client;
2+
3+
namespace EventMesh.Runtime.AMQP
4+
{
5+
public class AMQPSubscriptionRecord
6+
{
7+
public AMQPSubscriptionRecord(string topicName, string brokerName, string clientId, string clientSessionId, IModel channel, string consumerTag)
8+
{
9+
TopicName = topicName;
10+
BrokerName = brokerName;
11+
ClientId = clientId;
12+
ClientSessionId = clientSessionId;
13+
Channel = channel;
14+
ConsumerTag = consumerTag;
15+
}
16+
17+
public string TopicName { get; set; }
18+
public string BrokerName { get; set; }
19+
public string ClientId { get; set; }
20+
public string ClientSessionId { get; set; }
21+
public IModel Channel { get; set; }
22+
public string ConsumerTag { get; set; }
23+
}
24+
}

‎src/EventMesh.Runtime.EF/Stores/EFClientStore.cs‎

Lines changed: 36 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ namespace EventMesh.Runtime.EF.Stores
88
{
99
public class EFClientStore : IClientStore
1010
{
11+
private static object _obj = new object();
1112
private readonly EventMeshDBContext _dbContext;
1213

1314
public EFClientStore(EventMeshDBContext dbContext)
@@ -17,66 +18,65 @@ public EFClientStore(EventMeshDBContext dbContext)
1718

1819
public void Add(Client session)
1920
{
20-
_dbContext.Clients.Add(session);
21-
_dbContext.SaveChanges();
21+
lock (_obj)
22+
{
23+
_dbContext.Clients.Add(session);
24+
_dbContext.SaveChanges();
25+
}
2226
}
2327

2428
public int Count()
2529
{
26-
return _dbContext.Clients.Count();
30+
lock(_obj)
31+
return _dbContext.Clients.Count();
2732
}
2833

2934
public int CountActiveSessions()
3035
{
31-
return _dbContext.Clients.SelectMany(c => c.Sessions).Count(c => c.State == ClientSessionState.ACTIVE);
36+
lock(_obj)
37+
return _dbContext.Clients.SelectMany(c => c.Sessions).Count(c => c.State == ClientSessionState.ACTIVE);
3238
}
3339

3440
public Client Get(string clientId)
3541
{
36-
return _dbContext.Clients
37-
.Include(c => c.Sessions).ThenInclude(c => c.Histories)
38-
.Include(c => c.Sessions).ThenInclude(c => c.Topics)
39-
.Include(c => c.Sessions).ThenInclude(c => c.PendingCloudEvents)
40-
.Include(c => c.Sessions).ThenInclude(c => c.Bridges)
41-
.Include(c => c.Topics)
42-
.FirstOrDefault(c => c.ClientId == clientId);
42+
lock(_obj)
43+
return _dbContext.Clients
44+
.Include(c => c.Sessions).ThenInclude(c => c.Histories)
45+
.Include(c => c.Sessions).ThenInclude(c => c.Topics)
46+
.Include(c => c.Sessions).ThenInclude(c => c.PendingCloudEvents)
47+
.Include(c => c.Sessions).ThenInclude(c => c.Bridges)
48+
.Include(c => c.Topics)
49+
.FirstOrDefault(c => c.ClientId == clientId);
4350
}
4451

4552
public IEnumerable<Client> GetAll()
4653
{
47-
return _dbContext.Clients
48-
.Include(c => c.Sessions)
49-
.Include(c => c.Topics)
50-
.ToList();
51-
}
52-
53-
public IEnumerable<Client> GetAllBySubscribedTopics(string brokerName, string topicName)
54-
{
55-
return _dbContext.Clients
56-
.Include(c => c.Sessions).ThenInclude(c => c.Histories)
57-
.Include(c => c.Sessions).ThenInclude(c => c.Topics)
58-
.Include(c => c.Sessions).ThenInclude(c => c.PendingCloudEvents)
59-
.Include(c => c.Sessions).ThenInclude(c => c.Bridges)
60-
.Include(c => c.Topics)
61-
.Where(c => c.Sessions.Any(s => s.State == ClientSessionState.ACTIVE && s.Topics.Any(t => t.BrokerName == brokerName && t.Name == topicName)))
62-
.ToList();
54+
lock(_obj)
55+
return _dbContext.Clients
56+
.Include(c => c.Sessions)
57+
.Include(c => c.Topics)
58+
.ToList();
6359
}
6460

6561
public Client GetByActiveSession(string clientId, string sessionId)
6662
{
67-
return _dbContext.Clients
68-
.Include(c => c.Sessions).ThenInclude(c => c.Histories)
69-
.Include(c => c.Sessions).ThenInclude(c => c.Topics)
70-
.Include(c => c.Sessions).ThenInclude(c => c.PendingCloudEvents)
71-
.Include(c => c.Sessions).ThenInclude(c => c.Bridges)
72-
.Include(c => c.Topics)
73-
.FirstOrDefault(c => c.ClientId == clientId && c.Sessions.Any(s => s.Id == sessionId));
63+
lock(_obj)
64+
return _dbContext.Clients
65+
.Include(c => c.Sessions).ThenInclude(c => c.Histories)
66+
.Include(c => c.Sessions).ThenInclude(c => c.Topics)
67+
.Include(c => c.Sessions).ThenInclude(c => c.PendingCloudEvents)
68+
.Include(c => c.Sessions).ThenInclude(c => c.Bridges)
69+
.Include(c => c.Topics)
70+
.FirstOrDefault(c => c.ClientId == clientId && c.Sessions.Any(s => s.Id == sessionId));
7471
}
7572

7673
public void Update(Client client)
7774
{
78-
_dbContext.Clients.Update(client);
79-
_dbContext.SaveChanges();
75+
lock(_obj)
76+
{
77+
_dbContext.Clients.Update(client);
78+
_dbContext.SaveChanges();
79+
}
8080
}
8181
}
8282
}

‎src/EventMesh.Runtime/Handlers/DisconnectMessageHandler.cs‎

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
using EventMesh.Runtime.Messages;
22
using EventMesh.Runtime.Models;
33
using EventMesh.Runtime.Stores;
4+
using System.Collections.Generic;
5+
using System.Linq;
46
using System.Net;
57
using System.Threading;
68
using System.Threading.Tasks;
@@ -9,16 +11,16 @@ namespace EventMesh.Runtime.Handlers
911
{
1012
public class DisconnectMessageHandler : BaseMessageHandler, IMessageHandler
1113
{
12-
private readonly IBridgeServerStore _bridgeServerStore;
14+
private readonly IEnumerable<IMessageConsumer> _messageConsumers;
1315
private readonly IUdpClientServerFactory _udpClientFactory;
1416

1517
public DisconnectMessageHandler(
1618
IClientStore clientStore,
17-
IBridgeServerStore bridgeServerStore,
19+
IEnumerable<IMessageConsumer> messageConsumers,
1820
IUdpClientServerFactory udpClientServerFactory) : base(clientStore)
1921
{
20-
_bridgeServerStore = bridgeServerStore;
2122
_udpClientFactory = udpClientServerFactory;
23+
_messageConsumers = messageConsumers;
2224
}
2325

2426
public Commands Command => Commands.DISCONNECT_REQUEST;
@@ -28,12 +30,19 @@ public async Task<Package> Run(Package package, IPEndPoint sender, CancellationT
2830
var disconnectRequest = package as DisconnectRequest;
2931
var client = GetActiveSession(package, disconnectRequest.ClientId, disconnectRequest.SessionId);
3032
await CloseRemoteSessions(client, disconnectRequest);
31-
CloseLocalSession(client, disconnectRequest.SessionId);
33+
await CloseLocalSession(client, disconnectRequest.SessionId);
3234
return PackageResponseBuilder.Disconnect(package.Header.Seq);
3335
}
3436

35-
private void CloseLocalSession(Client client, string sessionId)
37+
private async Task CloseLocalSession(Client client, string sessionId)
3638
{
39+
var activeSession = client.GetActiveSession(sessionId);
40+
foreach (var topic in activeSession.Topics)
41+
{
42+
var messageConsumer = _messageConsumers.First(m => m.BrokerName == topic.BrokerName);
43+
await messageConsumer.Unsubscribe(topic.Name, client, sessionId, CancellationToken.None);
44+
}
45+
3746
client.CloseActiveSession(sessionId);
3847
ClientStore.Update(client);
3948
}

‎src/EventMesh.Runtime/IMessageConsumer.cs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ namespace EventMesh.Runtime
88
{
99
public interface IMessageConsumer : IDisposable
1010
{
11+
string BrokerName { get; }
1112
Task Start(CancellationToken cancellationToken);
1213
Task Stop(CancellationToken cancellationToken);
1314
Task Subscribe(string topic, Client client, string sessionId, CancellationToken cancellationToken);

‎src/EventMesh.Runtime/MessageBroker/InMemoryMessageConsumer.cs‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,14 @@ public InMemoryMessageConsumer(ICollection<InMemoryTopic> topics)
1818
_topics = topics;
1919
}
2020

21+
public string BrokerName
22+
{
23+
get
24+
{
25+
return Constants.InMemoryBrokername;
26+
}
27+
}
28+
2129
public Task Start(CancellationToken cancellationToken)
2230
{
2331
return Task.CompletedTask;

‎src/EventMesh.Runtime/Models/ClientSession.cs‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -82,12 +82,6 @@ public void SubscribeTopic(string topicName, string brokerName)
8282
});
8383
}
8484

85-
public void UnsubscribeTopic(string topicName, string brokerName)
86-
{
87-
var topic = Topics.First(t => t.Name == topicName && t.BrokerName == brokerName);
88-
Topics.Remove(topic);
89-
}
90-
9185
public bool HasTopic(string topicName, string brokerName)
9286
{
9387
return Topics.Any(t => t.Name == topicName && t.BrokerName == brokerName);

‎src/EventMesh.Runtime/RuntimeClient.cs‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -281,15 +281,13 @@ private async Task Run(CancellationToken cancellationToken)
281281
{
282282
receiveResult = await _udpClient.ReceiveAsync().WithCancellation(cancellationToken);
283283
}
284-
catch (Exception ex)
284+
catch (Exception)
285285
{
286-
Console.WriteLine(ex.ToString());
287286
return;
288287
}
289288

290289
var buffer = receiveResult.Buffer;
291290
var package = Package.Deserialize(new ReadBufferContext(buffer));
292-
Console.WriteLine(package.Header.Command.Code);
293291
if (package.Header.Command == Commands.ASYNC_MESSAGE_TO_CLIENT)
294292
{
295293
var asyncMessage = package as AsyncMessageToClient;

‎src/EventMesh.Runtime/RuntimeHost.cs‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -172,7 +172,6 @@ private async Task HandleCloudEventReceived(object sender, CloudEventArgs e)
172172
PackageResponseBuilder.AsyncMessageToServer(e.ClientId, bridgeServers, e.BrokerName, e.Topic, pendingCloudEvts, e.ClientSession.Id).Serialize(writeCtx);
173173
break;
174174
case Models.ClientSessionTypes.CLIENT:
175-
Console.WriteLine("Send to client");
176175
PackageResponseBuilder.AsyncMessageToClient(bridgeServers, e.BrokerName, e.Topic, pendingCloudEvts).Serialize(writeCtx);
177176
break;
178177
}

‎src/EventMesh.Runtime/Stores/ClientStore.cs‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -54,10 +54,5 @@ public IEnumerable<Client> GetAll()
5454
{
5555
return _clients;
5656
}
57-
58-
public IEnumerable<Client> GetAllBySubscribedTopics(string brokerName, string topicName)
59-
{
60-
return _clients.Where(c => c.ActiveSessions.Any(a => a.HasTopic(topicName, brokerName)));
61-
}
6257
}
6358
}

0 commit comments

Comments
 (0)