producer and cunsomer added

SocialMinistryDataExchange/Controllers/apiController.cs

@@ -2,6 +2,7 @@
 using SocialMinistryDataExchange.Model;
 using SocialMinistryDataExchange.Model;
 using System;
 using System;
 using System.Collections.Generic;
 using System.Collections.Generic;
+using System.Net;
 using System.Text.Json;
 using System.Text.Json;
 using System.Threading.Tasks;
 using System.Threading.Tasks;
 
 
@@ -9,9 +10,10 @@ namespace SocialMinistryDataExchange.Controllers {
     [ApiController]
     [ApiController]
     public class apiController : ControllerBase {
     public class apiController : ControllerBase {
         private readonly ISMRepository _repository;
         private readonly ISMRepository _repository;
-
-        public apiController(ISMRepository repository) { 
+        private readonly MessageProducer _messageProducer;
+        public apiController(ISMRepository repository, MessageProducer messageProducer) { 
             _repository = repository;
             _repository = repository;
+            _messageProducer = messageProducer;
         }
         }
 
 
         //тестовый запрос для симуляции получения XML запроса
         //тестовый запрос для симуляции получения XML запроса
@@ -23,6 +25,15 @@ namespace SocialMinistryDataExchange.Controllers {
             return responce;
             return responce;
         }
         }
 
 
+        //тестовый запрос для симуляции получения XML запроса
+        [HttpGet]
+        [Route("[controller]/putXMLToActiveMQServer")]
+        public async Task<IActionResult> PutXMLToAMQ() {
+            var request = System.IO.File.ReadAllText("Model/reference/Request.xml");
+            await _messageProducer.PublishAsync(request);
+            return StatusCode((int)HttpStatusCode.Created, null);
+        }
+
         // тестовый запрос зачисленности личности
         // тестовый запрос зачисленности личности
         [HttpGet]
         [HttpGet]
         [Route("[controller]/getFixPerson")]
         [Route("[controller]/getFixPerson")]

SocialMinistryDataExchange/Model/ActiveMqExtensions.cs

+using ActiveMQ.Artemis.Client;
+using ActiveMQ.Artemis.Client.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+using System;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace SocialMinistryDataExchange.Model {
+    public static class ActiveMqExtensions {
+        //public static IActiveMqBuilder AddTypedConsumer<TMessage, TConsumer>(this IActiveMqBuilder builder, RoutingType routingType) where TConsumer : class, ITypedConsumer<TMessage> {
+        public static IActiveMqBuilder AddTypedConsumer<TMessage, TConsumer>(this IActiveMqBuilder builder, string address, RoutingType routingType) where TConsumer : class, ITypedConsumer<TMessage> {
+            builder.Services.TryAddScoped<TConsumer>();
+            //builder.AddConsumer(typeof(TMessage).Name, routingType, HandleCustomerMessage<TMessage, TConsumer>);
+            builder.AddConsumer(address, routingType, HandleCustomerMessage<TMessage, TConsumer>);
+            return builder;
+        }
+        private static async Task HandleCustomerMessage<TMessage, TConsumer>(Message message, IConsumer consumer, IServiceProvider serviceProvider, CancellationToken token) where TConsumer : class, ITypedConsumer<TMessage> {
+            var msg = JsonSerializer.Deserialize<TMessage>(message.GetBody<string>());
+            using var scope = serviceProvider.CreateScope();
+            var typedConsumer = scope.ServiceProvider.GetService<TConsumer>();
+            await typedConsumer.ConsumeAsync(msg, token);
+            await consumer.AcceptAsync(message);
+        }
+    }
+}
\ No newline at end of file

SocialMinistryDataExchange/Model/ITypedConsumer.cs

+using System.Threading;
+using System.Threading.Tasks;
+
+namespace SocialMinistryDataExchange.Model {
+    public interface ITypedConsumer<in T> {
+        public Task ConsumeAsync(T message, CancellationToken cancellationToken);
+    }
+}
\ No newline at end of file

SocialMinistryDataExchange/Model/SMRepository.cs

@@ -32,8 +32,6 @@ namespace SocialMinistryDataExchange.Model {
             using (StringWriter writer = new()) {
             using (StringWriter writer = new()) {
                 xmlSerializer.Serialize(writer, request);
                 xmlSerializer.Serialize(writer, request);
                 responce = writer.ToString();
                 responce = writer.ToString();
-
-
             }
             }
             return Task.FromResult(responce);
             return Task.FromResult(responce);
         }
         }

SocialMinistryDataExchange/Model/SmMessageCunsomers.cs

+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace SocialMinistryDataExchange.Model {
+    public class StringMessageCunsomer : ITypedConsumer<String> {
+        public StringMessageCunsomer() {
+        }
+        public async Task ConsumeAsync(String message, CancellationToken cancellationToken) {
+            // тут логика
+            Console.WriteLine(message);
+        }
+    }
+}

SocialMinistryDataExchange/Model/SmMessageProdusers.cs

+using ActiveMQ.Artemis.Client;
+using SocialMinistryDataExchange.Model.EscpSD;
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Xml.Serialization;
+
+namespace SocialMinistryDataExchange.Model {
+    public class MessageProducer {
+        private readonly IProducer _producer;
+
+        public MessageProducer(IProducer producer) {
+            _producer = producer;
+        }
+        public async Task PublishAsync<T>(T message) {
+            XmlSerializer xmlSerializer = new(typeof(EcspSendingStatusMaloimushch));
+            string serialized = message.ToString();
+            //string serialized = string.Empty;
+            //using (StringWriter writer = new()) {
+            //    xmlSerializer.Serialize(writer, message);
+            //    serialized = writer.ToString();
+            //}
+
+            var msg = new Message(serialized);
+
+            try {
+                await _producer.SendAsync(msg);
+            }
+            catch (Exception ex) {
+            }
+        }
+    }
+}

SocialMinistryDataExchange/SocialMinistryDataExchange.csproj

@@ -5,6 +5,8 @@
   </PropertyGroup>
   </PropertyGroup>
 
 
   <ItemGroup>
   <ItemGroup>
+    <PackageReference Include="ArtemisNetClient" Version="2.7.0" />
+    <PackageReference Include="ArtemisNetClient.Extensions.DependencyInjection" Version="2.7.0" />
     <PackageReference Include="Serilog.Extensions.Logging.File" Version="2.0.0" />
     <PackageReference Include="Serilog.Extensions.Logging.File" Version="2.0.0" />
     <PackageReference Include="Swashbuckle.AspNetCore" Version="5.6.3" />
     <PackageReference Include="Swashbuckle.AspNetCore" Version="5.6.3" />
     <PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="5.0.9" />
     <PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="5.0.9" />

SocialMinistryDataExchange/Startup.cs

+using ActiveMQ.Artemis.Client;
+using ActiveMQ.Artemis.Client.Extensions.DependencyInjection;
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.EntityFrameworkCore;
@@ -41,6 +43,15 @@ namespace SocialMinistryDataExchange {
             services.AddDbContext<Contingent_center_Context>(options => options.UseSqlServer(ConnectionString));
             services.AddDbContext<Contingent_center_Context>(options => options.UseSqlServer(ConnectionString));
             services.AddTransient<IContingentContext, Contingent_center_Context>();
             services.AddTransient<IContingentContext, Contingent_center_Context>();
             services.AddTransient<ISMRepository, SMRepository>();
             services.AddTransient<ISMRepository, SMRepository>();
+
+            string SmQueue = "MV.SMEV_INF_DAN_OBE_ZHIL.REQ";
+            try {
+                services.AddActiveMq("ddoApp-cluster", new[] { Endpoint.Create(host: "172.17.100.121", port: 61616, "", "") })
+                    .AddProducer<MessageProducer>(SmQueue, RoutingType.Anycast)
+                    .AddTypedConsumer<string, StringMessageCunsomer>(SmQueue, RoutingType.Anycast);
+            }
+            catch (Exception ex) {
+            }
         }
         }
 
 
         // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
         // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.