Skip to content
This repository was archived by the owner on Nov 17, 2023. It is now read-only.

Commit c357aea

Browse files
zedy-wjzedy-wj
authored
Update ServiceBus sdk from t1 to t2 (#1698)
* Update ServiceBus to the new SDKs. * resolve errors * resolve errors * update codes * update codes * add a call in constructor * remove changes * add Dispose() menthod in interface * Update codes * update the method to get topicName * Remove invalid changes * resolve the errors in service * fix issue * fix issue * update code Co-authored-by: zedy <[email protected]>
1 parent 4621995 commit c357aea

File tree

13 files changed

+2145
-414
lines changed

13 files changed

+2145
-414
lines changed

‎src/BuildingBlocks/EventBus/EventBusServiceBus/DefaultServiceBusPersisterConnection.cs‎

Lines changed: 17 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -1,58 +1,49 @@
1-
using Microsoft.Azure.ServiceBus;
1+
using Azure.Messaging.ServiceBus;
2+
using Azure.Messaging.ServiceBus.Administration;
23
using System;
34

45
namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus
56
{
67
public class DefaultServiceBusPersisterConnection : IServiceBusPersisterConnection
78
{
8-
private readonly ServiceBusConnectionStringBuilder _serviceBusConnectionStringBuilder;
9-
private readonly string _subscriptionClientName;
10-
private SubscriptionClient _subscriptionClient;
11-
private ITopicClient _topicClient;
9+
private readonly string _serviceBusConnectionString;
10+
private ServiceBusClient _topicClient;
11+
private ServiceBusAdministrationClient _subscriptionClient;
1212

1313
bool _disposed;
1414

15-
public DefaultServiceBusPersisterConnection(ServiceBusConnectionStringBuilder serviceBusConnectionStringBuilder,
16-
string subscriptionClientName)
15+
public DefaultServiceBusPersisterConnection(string serviceBusConnectionString)
1716
{
18-
_serviceBusConnectionStringBuilder = serviceBusConnectionStringBuilder ??
19-
throw new ArgumentNullException(nameof(serviceBusConnectionStringBuilder));
20-
_subscriptionClientName = subscriptionClientName;
21-
_subscriptionClient = new SubscriptionClient(_serviceBusConnectionStringBuilder, subscriptionClientName);
22-
_topicClient = new TopicClient(_serviceBusConnectionStringBuilder, RetryPolicy.Default);
17+
_serviceBusConnectionString = serviceBusConnectionString;
18+
_subscriptionClient = new ServiceBusAdministrationClient(_serviceBusConnectionString);
19+
_topicClient = new ServiceBusClient(_serviceBusConnectionString);
2320
}
2421

25-
public ITopicClient TopicClient
22+
public ServiceBusClient TopicClient
2623
{
2724
get
2825
{
29-
if (_topicClient.IsClosedOrClosing)
26+
if (_topicClient.IsClosed)
3027
{
31-
_topicClient = new TopicClient(_serviceBusConnectionStringBuilder, RetryPolicy.Default);
28+
_topicClient = new ServiceBusClient(_serviceBusConnectionString);
3229
}
3330
return _topicClient;
3431
}
3532
}
3633

37-
public ISubscriptionClient SubscriptionClient
34+
public ServiceBusAdministrationClient AdministrationClient
3835
{
3936
get
4037
{
41-
if (_subscriptionClient.IsClosedOrClosing)
42-
{
43-
_subscriptionClient = new SubscriptionClient(_serviceBusConnectionStringBuilder, _subscriptionClientName);
44-
}
4538
return _subscriptionClient;
4639
}
4740
}
4841

49-
public ServiceBusConnectionStringBuilder ServiceBusConnectionStringBuilder => _serviceBusConnectionStringBuilder;
50-
51-
public ITopicClient CreateModel()
42+
public ServiceBusClient CreateModel()
5243
{
53-
if (_topicClient.IsClosedOrClosing)
44+
if (_topicClient.IsClosed)
5445
{
55-
_topicClient = new TopicClient(_serviceBusConnectionStringBuilder, RetryPolicy.Default);
46+
_topicClient = new ServiceBusClient(_serviceBusConnectionString);
5647
}
5748

5849
return _topicClient;
@@ -63,6 +54,7 @@ public void Dispose()
6354
if (_disposed) return;
6455

6556
_disposed = true;
57+
_topicClient.DisposeAsync().GetAwaiter().GetResult();
6658
}
6759
}
6860
}

‎src/BuildingBlocks/EventBus/EventBusServiceBus/EventBusServiceBus.cs‎

Lines changed: 56 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -1,35 +1,44 @@
1-
namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus
1+
using Azure.Messaging.ServiceBus;
2+
using Azure.Messaging.ServiceBus.Administration;
3+
using Autofac;
4+
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus;
5+
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
6+
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Events;
7+
using Microsoft.Extensions.Logging;
8+
using System;
9+
using System.Text.Json;
10+
using System.Threading.Tasks;
11+
using System.Text;
12+
13+
namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus
214
{
3-
using Autofac;
4-
using Microsoft.Azure.ServiceBus;
5-
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus;
6-
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
7-
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Events;
8-
using Microsoft.Extensions.Logging;
9-
using System;
10-
using System.Text;
11-
using System.Text.Json;
12-
using System.Threading.Tasks;
13-
14-
public class EventBusServiceBus : IEventBus
15+
public class EventBusServiceBus : IEventBus, IDisposable
1516
{
1617
private readonly IServiceBusPersisterConnection _serviceBusPersisterConnection;
1718
private readonly ILogger<EventBusServiceBus> _logger;
1819
private readonly IEventBusSubscriptionsManager _subsManager;
1920
private readonly ILifetimeScope _autofac;
21+
private readonly string _topicName = "eshop_event_bus";
22+
private readonly string _subscriptionName;
23+
private ServiceBusSender _sender;
24+
private ServiceBusProcessor _processor;
2025
private readonly string AUTOFAC_SCOPE_NAME = "eshop_event_bus";
2126
private const string INTEGRATION_EVENT_SUFFIX = "IntegrationEvent";
2227

2328
public EventBusServiceBus(IServiceBusPersisterConnection serviceBusPersisterConnection,
24-
ILogger<EventBusServiceBus> logger, IEventBusSubscriptionsManager subsManager, ILifetimeScope autofac)
29+
ILogger<EventBusServiceBus> logger, IEventBusSubscriptionsManager subsManager, ILifetimeScope autofac, string subscriptionClientName)
2530
{
2631
_serviceBusPersisterConnection = serviceBusPersisterConnection;
2732
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
2833
_subsManager = subsManager ?? new InMemoryEventBusSubscriptionsManager();
2934
_autofac = autofac;
35+
_subscriptionName = subscriptionClientName;
36+
_sender = _serviceBusPersisterConnection.TopicClient.CreateSender(_topicName);
37+
ServiceBusProcessorOptions options = new ServiceBusProcessorOptions { MaxConcurrentCalls = 10, AutoCompleteMessages = false };
38+
_processor = _serviceBusPersisterConnection.TopicClient.CreateProcessor(_topicName, _subscriptionName, options);
3039

3140
RemoveDefaultRule();
32-
RegisterSubscriptionClientMessageHandler();
41+
RegisterSubscriptionClientMessageHandlerAsync().GetAwaiter().GetResult();
3342
}
3443

3544
public void Publish(IntegrationEvent @event)
@@ -38,14 +47,14 @@ public void Publish(IntegrationEvent @event)
3847
var jsonMessage = JsonSerializer.Serialize(@event, @event.GetType());
3948
var body = Encoding.UTF8.GetBytes(jsonMessage);
4049

41-
var message = new Message
50+
var message = new ServiceBusMessage
4251
{
4352
MessageId = Guid.NewGuid().ToString(),
44-
Body = body,
45-
Label = eventName,
53+
Body = new BinaryData(body),
54+
Subject = eventName,
4655
};
4756

48-
_serviceBusPersisterConnection.TopicClient.SendAsync(message)
57+
_sender.SendMessageAsync(message)
4958
.GetAwaiter()
5059
.GetResult();
5160
}
@@ -69,9 +78,9 @@ public void Subscribe<T, TH>()
6978
{
7079
try
7180
{
72-
_serviceBusPersisterConnection.SubscriptionClient.AddRuleAsync(new RuleDescription
81+
_serviceBusPersisterConnection.AdministrationClient.CreateRuleAsync(_topicName, _subscriptionName, new CreateRuleOptions
7382
{
74-
Filter = new CorrelationFilter { Label = eventName },
83+
Filter = new CorrelationRuleFilter() { Subject = eventName },
7584
Name = eventName
7685
}).GetAwaiter().GetResult();
7786
}
@@ -95,12 +104,12 @@ public void Unsubscribe<T, TH>()
95104
try
96105
{
97106
_serviceBusPersisterConnection
98-
.SubscriptionClient
99-
.RemoveRuleAsync(eventName)
107+
.AdministrationClient
108+
.DeleteRuleAsync(_topicName, _subscriptionName, eventName)
100109
.GetAwaiter()
101110
.GetResult();
102111
}
103-
catch (MessagingEntityNotFoundException)
112+
catch (ServiceBusException ex) when (ex.Reason == ServiceBusFailureReason.MessagingEntityNotFound)
104113
{
105114
_logger.LogWarning("The messaging entity {eventName} Could not be found.", eventName);
106115
}
@@ -118,32 +127,35 @@ public void UnsubscribeDynamic<TH>(string eventName)
118127
_subsManager.RemoveDynamicSubscription<TH>(eventName);
119128
}
120129

121-
public void Dispose()
122-
{
123-
_subsManager.Clear();
124-
}
125-
126-
private void RegisterSubscriptionClientMessageHandler()
130+
private async Task RegisterSubscriptionClientMessageHandlerAsync()
127131
{
128-
_serviceBusPersisterConnection.SubscriptionClient.RegisterMessageHandler(
129-
async (message, token) =>
132+
_processor.ProcessMessageAsync +=
133+
async (args) =>
130134
{
131-
var eventName = $"{message.Label}{INTEGRATION_EVENT_SUFFIX}";
132-
var messageData = Encoding.UTF8.GetString(message.Body);
135+
var eventName = $"{args.Message.Subject}{INTEGRATION_EVENT_SUFFIX}";
136+
string messageData = args.Message.Body.ToString();
133137

134138
// Complete the message so that it is not received again.
135139
if (await ProcessEvent(eventName, messageData))
136140
{
137-
await _serviceBusPersisterConnection.SubscriptionClient.CompleteAsync(message.SystemProperties.LockToken);
141+
await args.CompleteMessageAsync(args.Message);
138142
}
139-
},
140-
new MessageHandlerOptions(ExceptionReceivedHandler) { MaxConcurrentCalls = 10, AutoComplete = false });
143+
};
144+
145+
_processor.ProcessErrorAsync += ErrorHandler;
146+
await _processor.StartProcessingAsync();
147+
}
148+
149+
public void Dispose()
150+
{
151+
_subsManager.Clear();
152+
_processor.CloseAsync().GetAwaiter().GetResult();
141153
}
142154

143-
private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs)
155+
private Task ErrorHandler(ProcessErrorEventArgs args)
144156
{
145-
var ex = exceptionReceivedEventArgs.Exception;
146-
var context = exceptionReceivedEventArgs.ExceptionReceivedContext;
157+
var ex = args.Exception;
158+
var context = args.ErrorSource;
147159

148160
_logger.LogError(ex, "ERROR handling message: {ExceptionMessage} - Context: {@ExceptionContext}", ex.Message, context);
149161

@@ -189,14 +201,14 @@ private void RemoveDefaultRule()
189201
try
190202
{
191203
_serviceBusPersisterConnection
192-
.SubscriptionClient
193-
.RemoveRuleAsync(RuleDescription.DefaultRuleName)
204+
.AdministrationClient
205+
.DeleteRuleAsync(_topicName, _subscriptionName, RuleProperties.DefaultRuleName)
194206
.GetAwaiter()
195207
.GetResult();
196208
}
197-
catch (MessagingEntityNotFoundException)
209+
catch (ServiceBusException ex) when (ex.Reason == ServiceBusFailureReason.MessagingEntityNotFound)
198210
{
199-
_logger.LogWarning("The messaging entity {DefaultRuleName} Could not be found.", RuleDescription.DefaultRuleName);
211+
_logger.LogWarning("The messaging entity {DefaultRuleName} Could not be found.", RuleProperties.DefaultRuleName);
200212
}
201213
}
202214
}

‎src/BuildingBlocks/EventBus/EventBusServiceBus/EventBusServiceBus.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77

88
<ItemGroup>
99
<PackageReference Include="Autofac" Version="6.1.0" />
10-
<PackageReference Include="Microsoft.Azure.ServiceBus" Version="5.1.0" />
10+
<PackageReference Include="Azure.Messaging.ServiceBus" Version="7.2.1" />
1111
<PackageReference Include="Microsoft.CSharp" Version="4.7.0" />
1212
<PackageReference Include="Microsoft.Extensions.Logging" Version="5.0.0" />
1313
</ItemGroup>

‎src/BuildingBlocks/EventBus/EventBusServiceBus/IServiceBusPersisterConnection.cs‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,12 @@
1-
namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus
1+
using Azure.Messaging.ServiceBus;
2+
using Azure.Messaging.ServiceBus.Administration;
3+
namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus
24
{
3-
using Microsoft.Azure.ServiceBus;
45
using System;
56

67
public interface IServiceBusPersisterConnection : IDisposable
78
{
8-
ITopicClient TopicClient { get; }
9-
ISubscriptionClient SubscriptionClient { get; }
9+
ServiceBusClient TopicClient { get; }
10+
ServiceBusAdministrationClient AdministrationClient { get; }
1011
}
1112
}

‎src/Services/Basket/Basket.API/Startup.cs‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
using Autofac;
22
using Autofac.Extensions.DependencyInjection;
3+
using Azure.Messaging.ServiceBus;
34
using Basket.API.Infrastructure.Filters;
45
using Basket.API.IntegrationEvents.EventHandling;
56
using Basket.API.IntegrationEvents.Events;
@@ -10,7 +11,6 @@
1011
using Microsoft.AspNetCore.Diagnostics.HealthChecks;
1112
using Microsoft.AspNetCore.Hosting;
1213
using Microsoft.AspNetCore.Http;
13-
using Microsoft.Azure.ServiceBus;
1414
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus;
1515
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
1616
using Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ;
@@ -122,10 +122,8 @@ public virtual IServiceProvider ConfigureServices(IServiceCollection services)
122122
services.AddSingleton<IServiceBusPersisterConnection>(sp =>
123123
{
124124
var serviceBusConnectionString = Configuration["EventBusConnection"];
125-
var serviceBusConnection = new ServiceBusConnectionStringBuilder(serviceBusConnectionString);
126125

127-
var subscriptionClientName = Configuration["SubscriptionClientName"];
128-
return new DefaultServiceBusPersisterConnection(serviceBusConnection, subscriptionClientName);
126+
return new DefaultServiceBusPersisterConnection(serviceBusConnectionString);
129127
});
130128
}
131129
else
@@ -285,9 +283,10 @@ private void RegisterEventBus(IServiceCollection services)
285283
var iLifetimeScope = sp.GetRequiredService<ILifetimeScope>();
286284
var logger = sp.GetRequiredService<ILogger<EventBusServiceBus>>();
287285
var eventBusSubcriptionsManager = sp.GetRequiredService<IEventBusSubscriptionsManager>();
286+
string subscriptionName = Configuration["SubscriptionClientName"];
288287

289288
return new EventBusServiceBus(serviceBusPersisterConnection, logger,
290-
eventBusSubcriptionsManager, iLifetimeScope);
289+
eventBusSubcriptionsManager, iLifetimeScope, subscriptionName);
291290
});
292291
}
293292
else

‎src/Services/Catalog/Catalog.API/Startup.cs‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
using Autofac;
22
using Autofac.Extensions.DependencyInjection;
3+
using Azure.Messaging.ServiceBus;
34
using Catalog.API.Grpc;
45
using global::Catalog.API.Infrastructure.Filters;
56
using global::Catalog.API.IntegrationEvents;
@@ -9,7 +10,6 @@
910
using Microsoft.AspNetCore.Hosting;
1011
using Microsoft.AspNetCore.Http;
1112
using Microsoft.AspNetCore.Mvc;
12-
using Microsoft.Azure.ServiceBus;
1313
using Microsoft.EntityFrameworkCore;
1414
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus;
1515
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
@@ -281,10 +281,9 @@ public static IServiceCollection AddIntegrationServices(this IServiceCollection
281281
services.AddSingleton<IServiceBusPersisterConnection>(sp =>
282282
{
283283
var settings = sp.GetRequiredService<IOptions<CatalogSettings>>().Value;
284-
var serviceBusConnection = new ServiceBusConnectionStringBuilder(settings.EventBusConnection);
285-
var subscriptionClientName = configuration["SubscriptionClientName"];
284+
var serviceBusConnection = settings.EventBusConnection;
286285

287-
return new DefaultServiceBusPersisterConnection(serviceBusConnection, subscriptionClientName);
286+
return new DefaultServiceBusPersisterConnection(serviceBusConnection);
288287
});
289288
}
290289
else
@@ -333,9 +332,10 @@ public static IServiceCollection AddEventBus(this IServiceCollection services, I
333332
var iLifetimeScope = sp.GetRequiredService<ILifetimeScope>();
334333
var logger = sp.GetRequiredService<ILogger<EventBusServiceBus>>();
335334
var eventBusSubcriptionsManager = sp.GetRequiredService<IEventBusSubscriptionsManager>();
335+
string subscriptionName = configuration["SubscriptionClientName"];
336336

337337
return new EventBusServiceBus(serviceBusPersisterConnection, logger,
338-
eventBusSubcriptionsManager, iLifetimeScope);
338+
eventBusSubcriptionsManager, iLifetimeScope, subscriptionName);
339339
});
340340

341341
}

‎src/Services/Ordering/Ordering.API/Ordering.API.csproj‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@
6969
<PackageReference Include="Swashbuckle.AspNetCore" Version="5.6.3" />
7070
</ItemGroup>
7171

72+
7273
<ItemGroup>
7374
<None Update="Setup\*">
7475
<CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>

0 commit comments

Comments
 (0)