added k8s config

RabbitMQ/ConsumeRabbitMQHostedService.cs

@@ -1,68 +0,0 @@
-using RabbitMQ.Client;
-using RabbitMQ.Client.Events;
-using Microsoft.Extensions.Hosting;
-using System;
-using System.Threading;
-using Microsoft.Extensions.Logging;
-using System.Threading.Tasks;
-
-namespace OutputServiceTSDB.RabbitMQ
-{
-    public class ConsumeRabbitMQHostedService : BackgroundService
-    {
-        private readonly ILogger _logger;
-        private IConnection _connection;
-        private IModel _channel;
-
-        public ConsumeRabbitMQHostedService(ILoggerFactory loggerFactory)
-        {
-            this._logger = loggerFactory.CreateLogger<ConsumeRabbitMQHostedService>();
-            InitRabbitMQ();
-        }
-
-        private void InitRabbitMQ()
-        {
-            var factory = new ConnectionFactory { HostName = "localhost" };
-
-            _connection = factory.CreateConnection();
-
-            _channel = _connection.CreateModel();
-
-            _channel.ExchangeDeclare("demo.exchange", ExchangeType.Topic);
-            _channel.QueueDeclare("demo.queue.log", false, false, false, null);
-            _channel.QueueBind("demo.queue.log", "demo.exchange", "demo.queue.*", null);
-            _channel.BasicQos(0, 1, false);
-
-            _connection.ConnectionShutdown += RabbitMQ_ConnectionShutdown;
-        }
-
-        protected override Task ExecuteAsync(CancellationToken stoppingToken)
-        {
-            stoppingToken.ThrowIfCancellationRequested();
-
-            var consumer = new EventingBasicConsumer(_channel);
-            consumer.Received += (ch, ea) =>
-            {
-                var content = System.Text.Encoding.UTF8.GetString(ea.Body);
-
-                HandleMessage(content);
-                _channel.BasicAck(ea.DeliveryTag, false);
-            };
-
-            _channel.BasicConsume("demo.queue.log", false, consumer);
-            return Task.CompletedTask;
-        }
-
-        private void HandleMessage(string content)
-        {
-            _logger.LogInformation($"consumer received {content}");
-        }
-
-        public override void Dispose()
-        {
-            _channel.Close();
-            _connection.Close();
-            base.Dispose();
-        }
-    }
-}

RabbitMQ/RabbitMQWorker.cs

@@ -0,0 +1,64 @@
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+using OutputServiceTSDB.Utilities;
+using Microsoft.Extensions.Logging;
+using System.Threading.Tasks;
+using OutputServiceTSDB.InfluxDB;
+using OutputServiceTSDB.Models;
+using Newtonsoft.Json;
+using System;
+
+namespace OutputServiceTSDB.RabbitMQ
+{
+    public class RabbitMQWorker : IDisposable
+    {
+        private readonly ILogger _logger;
+        private IConnection _connection;
+        private IModel _channel;
+        private InfluxWriter influxWriter;
+
+        public RabbitMQWorker(ILogger<RabbitMQWorker> logger)
+        {
+            _logger = logger;
+            influxWriter = new InfluxWriter();
+            InitRabbitMQ();
+        }
+
+        private void InitRabbitMQ()
+        {
+            var factory = new ConnectionFactory { HostName = EnvironmentVariableConfiguration.RabbitMQHostname, UserName = EnvironmentVariableConfiguration.RabbitMQUserName, Password = EnvironmentVariableConfiguration.RabbitMQPassword };
+
+            _connection = factory.CreateConnection();
+
+            _channel = _connection.CreateModel();
+
+            _channel.ExchangeDeclare(EnvironmentVariableConfiguration.RabbitMQExchange, ExchangeType.Topic);
+            _channel.QueueDeclare(EnvironmentVariableConfiguration.RabbitMQQueue, false, false, false, null);
+            _channel.QueueBind(EnvironmentVariableConfiguration.RabbitMQQueue, EnvironmentVariableConfiguration.RabbitMQExchange, null, null);
+            _channel.BasicQos(0, 1, false);
+        }
+
+        public void ProcessMessages()
+        {
+            var consumer = new AsyncEventingBasicConsumer(_channel);
+            consumer.Received += HandleMessage;
+
+            _channel.BasicConsume("demo.queue.log", false, consumer);
+        }
+
+        private async Task HandleMessage(object sender, BasicDeliverEventArgs @event)
+        {
+            var content = System.Text.Encoding.UTF8.GetString(@event.Body);
+            _logger.LogInformation($"received {content}");
+            MeasurementObject measurementObject = JsonConvert.DeserializeObject<MeasurementObject>(content);
+            await Task.Run(() => influxWriter.Write(measurementObject));
+        }
+
+        public void Dispose()
+        {
+            _channel.Close();
+            _connection.Close();
+            influxWriter.Dispose();
+        }
+    }
+}