Polub 0
Zforkuj 0

Ostatnio aktywny 1725699790

Rewizja 5cfc4f09b4313a12ed53992f3437d233f159b3ae

Program.cs Surowy
1using System.Text;
2using Azure.Identity;
3using Azure.Messaging.ServiceBus;
4using Microsoft.Extensions.DependencyInjection;
5using Microsoft.Extensions.Logging;
6using Microsoft.Identity.Client;
7using Newtonsoft.Json;
8using LogLevel = Microsoft.Extensions.Logging.LogLevel;
9
10namespace LabServiceBus;
11
12public abstract class Program
13{
14 public static async Task Main(string[] args)
15 {
16 var services = new ServiceCollection();
17 services.AddLogging(logging =>
18 {
19 logging.AddConsole();
20 logging.SetMinimumLevel(LogLevel.Information);
21 });
22
23 services.AddHttpClient();
24 services.AddSingleton<ServiceBusEventSender>();
25
26 var serviceProvider = services.BuildServiceProvider();
27 var entry = serviceProvider.GetRequiredService<ServiceBusEventSender>();
28
29 await Task.Factory.StartNew(async () =>
30 {
31 try
32 {
33 await entry.Listen(async message =>
34 {
35 await Task.CompletedTask;
36 Console.WriteLine($"==================Received message: {message}==================");
37 });
38 }
39 catch (Exception e)
40 {
41 Console.WriteLine(e);
42 throw;
43 }
44 });
45
46 for (int i = 0; i < 1000; i++)
47 {
48 await entry.SendAsync(new TestEvent { Name = $"Test {i}" });
49 }
50
51 await Task.Delay(int.MaxValue);
52 }
53}
54
55public class TestEvent
56{
57 // ReSharper disable once UnusedAutoPropertyAccessor.Global
58 public string? Name { get; set; }
59}
60
61public class ServiceBusEventSender(ILogger<ServiceBusEventSender> logger, IHttpClientFactory httpClientFactory)
62{
63 private readonly HttpClient _httpClient = httpClientFactory.CreateClient();
64 private readonly string _tenantId = "538bf3b7-8deb-4179-9c63-f09d13f65838";
65 private readonly string _clientId = "a19850a3-cfef-4ced-8fcd-d632fec31bdc";
66 private readonly string _clientSecret = "YOUR_CLIENT_SECRET";
67 private readonly string _scope = "https://servicebus.azure.net/.default";
68
69 private readonly string _serviceBusNamespace = "anduinlearn";
70 private readonly string _topicName = "mytopic";
71 private readonly string _subscriptionName = "mysubscription";
72
73 /// <summary>
74 /// Sends an asynchronous request to the service bus endpoint.
75 /// </summary>
76 /// <typeparam name="T">The type of the request object.</typeparam>
77 /// <param name="request">The request object.</param>
78 /// <returns>A task representing the asynchronous operation that returns the response string.</returns>
79 public async Task<string> SendAsync<T>(T request)
80 {
81 var requestJson = JsonConvert.SerializeObject(request);
82 logger.LogTrace("Sending request to service bus, with input {0}", request?.GetType().FullName);
83
84 var endpoint = $"https://{_serviceBusNamespace}.servicebus.windows.net/{_topicName}/messages";
85 logger.LogTrace("Sending request to service bus, with endpoint {0}", endpoint);
86
87 var token = await GetToken();
88 logger.LogTrace("Sending request to service bus, with token {0}", token.Substring(0, 20));
89
90 var httpRequest = new HttpRequestMessage
91 {
92 Method = HttpMethod.Post,
93 RequestUri = new Uri(endpoint)
94 };
95 httpRequest.Headers.Add("Authorization", $"Bearer {token}");
96 httpRequest.Headers.Add("accept", "application/json");
97 httpRequest.Content = new StringContent(requestJson, Encoding.UTF8, "application/json");
98
99 logger.LogTrace("Sending request to service bus, with content {0}", requestJson);
100 var response = await _httpClient.SendAsync(httpRequest);
101
102 logger.LogTrace("Received response from service bus, with status code {0}", response.StatusCode);
103 var responseString = await response.Content.ReadAsStringAsync();
104
105 logger.LogTrace("Received response from service bus, with content {0} and status code {1}", responseString,
106 response.StatusCode);
107 if (!response.IsSuccessStatusCode)
108 {
109 throw new Exception($"Failed to send new service bus event. {responseString}");
110 }
111
112 return responseString;
113 }
114
115 public async Task Listen(Func<string, Task> onMessage)
116 {
117 // Create Service Bus client and processor
118 logger.LogWarning("Listening to service bus topic {0} and subscription {1}", _topicName, _subscriptionName);
119 var credential = new ClientSecretCredential(_tenantId, _clientId, _clientSecret);
120 var serviceBusClient = new ServiceBusClient(_serviceBusNamespace + ".servicebus.windows.net"
121 , credential);
122 var processor = serviceBusClient.CreateProcessor(_topicName, _subscriptionName, new ServiceBusProcessorOptions());
123
124 processor.ProcessMessageAsync += async args =>
125 {
126 var message = args.Message;
127 var body = Encoding.UTF8.GetString(message.Body.ToArray());
128 await onMessage(body);
129 await args.CompleteMessageAsync(message);
130 };
131
132 processor.ProcessErrorAsync += ErrorHandler;
133
134 logger.LogWarning("Starting to listen to service bus topic {0} and subscription {1}", _topicName, _subscriptionName);
135 // Start processing
136 await processor.StartProcessingAsync();
137 await Task.Delay(int.MaxValue);
138 }
139
140 private Task ErrorHandler(ProcessErrorEventArgs args)
141 {
142 logger.LogError(args.Exception, "Error processing message: {ExceptionMessage}", args.Exception.Message);
143 return Task.CompletedTask;
144 }
145
146 private async Task<string> GetToken()
147 {
148 var app = ConfidentialClientApplicationBuilder.Create(_clientId)
149 .WithClientSecret(_clientSecret)
150 .WithAuthority(new Uri($"https://login.microsoftonline.com/{_tenantId}"))
151 .Build();
152
153 var authResult = await app.AcquireTokenForClient(new[] { _scope }).ExecuteAsync();
154 return authResult.AccessToken;
155 }
156}
157

Obsługiwane przez Opengist ⋅ Load: 10ms⋅

Polski