Merge pull request 'greenhouseCRUD' (#7) from greenhouseCRUD into dev

Reviewed-on: #7

.dockerignore

@@ -0,0 +1,25 @@
+**/.classpath
+**/.dockerignore
+**/.env
+**/.git
+**/.gitignore
+**/.project
+**/.settings
+**/.toolstarget
+**/.vs
+**/.vscode
+**/*.*proj.user
+**/*.dbmdl
+**/*.jfm
+**/bin
+**/charts
+**/docker-compose*
+**/compose*
+**/Dockerfile*
+**/node_modules
+**/npm-debug.log
+**/obj
+**/secrets.dev.yaml
+**/values.dev.yaml
+LICENSE
+README.md

.env

@@ -0,0 +1,4 @@
+POSTGRES_USER="postgres"
+POSTGRES_PASSWORD="12345"
+POSTGRES_DB="main_database"
+DB_CONNECTION_STRING="Host=postgres:5432;Database=${POSTGRES_DB};Username=${POSTGRES_USER};Password=${POSTGRES_PASSWORD}"

.vscode/launch.json

@@ -0,0 +1,35 @@
+{
+    "version": "0.2.0",
+    "configurations": [
+        {
+            // Use IntelliSense to find out which attributes exist for C# debugging
+            // Use hover for the description of the existing attributes
+            // For further information visit https://github.com/dotnet/vscode-csharp/blob/main/debugger-launchjson.md.
+            "name": ".NET Core Launch (web)",
+            "type": "coreclr",
+            "request": "launch",
+            "preLaunchTask": "build",
+            // If you have changed target frameworks, make sure to update the program path.
+            "program": "${workspaceFolder}/Cloud/bin/Debug/net6.0/Cloud.dll",
+            "args": [],
+            "cwd": "${workspaceFolder}/Cloud",
+            "stopAtEntry": false,
+            // Enable launching a web browser when ASP.NET Core starts. For more information: https://aka.ms/VSCode-CS-LaunchJson-WebBrowser
+            "serverReadyAction": {
+                "action": "openExternally",
+                "pattern": "\\bNow listening on:\\s+(https?://\\S+)"
+            },
+            "env": {
+                "ASPNETCORE_ENVIRONMENT": "Development"
+            },
+            "sourceFileMap": {
+                "/Views": "${workspaceFolder}/Views"
+            }
+        },
+        {
+            "name": ".NET Core Attach",
+            "type": "coreclr",
+            "request": "attach"
+        }
+    ]
+}

.vscode/tasks.json

@@ -0,0 +1,41 @@
+{
+    "version": "2.0.0",
+    "tasks": [
+        {
+            "label": "build",
+            "command": "dotnet",
+            "type": "process",
+            "args": [
+                "build",
+                "${workspaceFolder}/Cloud.sln",
+                "/property:GenerateFullPaths=true",
+                "/consoleloggerparameters:NoSummary;ForceNoAlign"
+            ],
+            "problemMatcher": "$msCompile"
+        },
+        {
+            "label": "publish",
+            "command": "dotnet",
+            "type": "process",
+            "args": [
+                "publish",
+                "${workspaceFolder}/Cloud.sln",
+                "/property:GenerateFullPaths=true",
+                "/consoleloggerparameters:NoSummary;ForceNoAlign"
+            ],
+            "problemMatcher": "$msCompile"
+        },
+        {
+            "label": "watch",
+            "command": "dotnet",
+            "type": "process",
+            "args": [
+                "watch",
+                "run",
+                "--project",
+                "${workspaceFolder}/Cloud.sln"
+            ],
+            "problemMatcher": "$msCompile"
+        }
+    ]
+}

Cloud/ApplicationContext.cs

@@ -6,6 +6,7 @@ public class ApplicationContext : DbContext
 {
     public DbSet<User> Users { get; set; } = null!;
     public DbSet<Farm> Farms { get; set; } = null!;
+    public DbSet<Greenhouse> Greenhouses { get; set; } = null!;
 
     public ApplicationContext(DbContextOptions<ApplicationContext> options)
         : base(options)

Cloud/Cloud.csproj

@@ -7,6 +7,7 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="Confluent.Kafka" Version="2.6.1" />
     <PackageReference Include="FluentValidation.AspNetCore" Version="11.3.0" />
     <PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="6.0.4" />
     <PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="6.0.14" />

Cloud/Controllers/FarmController.cs

@@ -21,7 +21,7 @@ namespace Cloud.Controllers
         }
 
         [HttpGet("{userId}/farm")]
-		public async Task<ActionResult<List<Farm>>> Index (int userId)
+        public async Task<ActionResult<List<Farm>>> Index(int userId)
         {
             try
             {
@@ -62,10 +62,11 @@ namespace Cloud.Controllers
         {
             try
             {
-				var farm = new Farm {
+                var farm = new Farm
+                {
                     Name = farmRequest.Name,
                     UserId = userId,
-					RaspberryMacAddr = farmRequest.RaspberryMacAddr,
+                    RaspberryIP = farmRequest.RaspberryIP,
                 };
 
                 Farm? farmCreated = _context.Farms.Add(farm).Entity;
@@ -90,7 +91,7 @@ namespace Cloud.Controllers
                     return NotFound("Farm is not found");
 
                 farm.Name = farmRequest.Name;
-				farm.RaspberryMacAddr = farmRequest.RaspberryMacAddr;
+                farm.RaspberryIP = farmRequest.RaspberryIP;
 
                 _context.Farms.Update(farm);
                 await _context.SaveChangesAsync();
@@ -123,6 +124,5 @@ namespace Cloud.Controllers
                 return BadRequest(ex.Message);
             }
         }
-
     }
 }

Cloud/Controllers/GreengouseController.cs

@@ -0,0 +1,160 @@
+using Cloud.Models;
+using Cloud.Requests;
+using Cloud.Services.Broker;
+using Cloud.Services.Broker.Support;
+using Cloud.Services.Domain;
+using Microsoft.AspNetCore.Mvc;
+
+namespace Cloud.Controllers
+{
+    [ApiController]
+    [Route("api/farm/{farmId}/greenhouse")]
+    public class GreenhouseController : ControllerBase 
+    {
+        private readonly IGreenhouseService _greenhouseService;
+        public GreenhouseController(IGreenhouseService greenhouseService) 
+        {
+            _greenhouseService = greenhouseService;
+        }
+
+        /// <summary>
+        /// Возвращает текущую информацию о всех теплицах пользователя
+        /// </summary>
+        /// <param name="farmId"></param>
+        /// <returns></returns>
+        [HttpGet]
+        public async Task<ActionResult<List<GreenhouseInfo>>> GetAll(int farmId) 
+        {
+            try
+            {
+                var greenhouses = _greenhouseService.GetAll(farmId);
+                if (greenhouses == null) return NotFound("Greenhouses is not found");
+                return Ok(greenhouses);
+            }
+            catch (Exception ex)
+            {
+                return BadRequest(ex.Message);
+            }
+        }
+
+        /// <summary>
+        /// Возвращает текущую информацию о конкретной теплице
+        /// </summary>
+        /// <param name="farmId"></param>
+        /// <param name="greenhouseId"></param>
+        /// <returns></returns>
+        [HttpGet("{greenhouseId}")]
+        public async Task<ActionResult<GreenhouseInfo>> Get(int farmId, int greenhouseId) 
+        {
+            try
+            {
+                var greenhouses = _greenhouseService.GetGreenhouseInfo(greenhouseId, farmId);
+                return Ok(greenhouses);
+            }
+            catch (Exception ex)
+            {
+                return BadRequest(ex.Message);
+            }
+        }
+
+        /// <summary>
+        /// Возвращает сохраненные данные для автоматизации теплицы
+        /// </summary>
+        /// <param name="farmId"></param>
+        /// <param name="greenhouseId"></param>
+        /// <returns></returns>
+        [HttpGet("{greenhouseId}/settings")]
+        public async Task<ActionResult<Greenhouse>> GetGreenhouse(int farmId, int greenhouseId) 
+        {
+            try 
+            {
+                var greenhouse = await _greenhouseService.GetGreenhouse(greenhouseId);
+                if (greenhouse == null) return NotFound("Greenhouses is not found");
+                return Ok(greenhouse);
+            }
+            catch (Exception ex)
+            {
+                return BadRequest(ex.Message);
+            }
+        }
+
+        /// <summary>
+        /// Сохраняет в базе данных API данные для автоматизации теплицы
+        /// </summary>
+        /// <param name="farmId"></param>
+        /// <param name="greenhouse"></param>
+        /// <returns></returns>
+        [HttpPost]
+        public async Task<ActionResult<Greenhouse>> SaveToDatabase(int farmId, GreenhouseRequest greenhouse) 
+        {
+            try
+            {
+                var greenhouseEntity = new Greenhouse() 
+                { 
+                    RecomendedTemperature = greenhouse.RecomendedTemperature,
+                    WateringMode = greenhouse.WateringMode,
+                    HeatingMode = greenhouse.HeatingMode
+                };
+
+                var result = await _greenhouseService.Create(greenhouseEntity);
+                return Ok(result);
+            }
+            catch (Exception ex)
+            {
+                return BadRequest(ex.Message);
+            }
+            
+        }
+
+        
+        /// <summary>
+        /// Обновляет в базе данных API данные для автоматизации теплицы
+        /// </summary>
+        /// <param name="farmId">ID фермы</param>
+        /// <param name="greenhouseId">ID теплицы</param>
+        /// <param name="greenhouse">Данные для обновления</param>
+        /// <returns>Обновленный объект Greenhouse</returns>
+        [HttpPut("{greenhouseId}/settings")]
+        public async Task<ActionResult<Greenhouse>> Update(int farmId, int greenhouseId, GreenhouseRequest greenhouse) 
+        {
+            try
+            {
+                var greenhouseEntity = new Greenhouse()
+                {
+                    Id = greenhouseId,
+                    FarmId = farmId,
+                    WateringMode = greenhouse.WateringMode,
+                    HeatingMode = greenhouse.HeatingMode,
+                    RecomendedTemperature = greenhouse.RecomendedTemperature
+                };
+
+                var result = await _greenhouseService.Update(greenhouseEntity);
+                return Ok(result);
+            }
+            catch (Exception ex)
+            {
+                return BadRequest(ex.Message);
+            }
+        }
+
+        /// <summary>
+        /// Удаляет из базы данных API запись настроек автоматизации теплицы
+        /// </summary>
+        /// <param name="farmId"></param>
+        /// <param name="greenhouseId"></param>
+        /// <returns></returns>
+        [HttpDelete("{greenhouseId}")]
+        public async Task<ActionResult> Delete(int farmId, int greenhouseId) 
+        {
+            try
+            {
+                _ = await _greenhouseService.Delete(greenhouseId);
+                return Ok();
+            }
+            catch (Exception ex)
+            {
+                return BadRequest(ex.Message);
+            }
+        }
+    }
+}

Cloud/Controllers/ValveController.cs

@@ -0,0 +1,43 @@
+using Cloud.Requests;
+using Cloud.Services;
+using Cloud.Services.Broker;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.Mvc;
+using System.ComponentModel.DataAnnotations;
+using System.Text.Json;
+
+namespace Cloud.Controllers
+{
+	[Authorize]
+	[ApiController]
+	[Route("api")]
+	public class ValveController : ControllerBase
+	{
+		//Контроллер вентиля
+
+		private readonly IBrokerService _kafkaService;
+
+		public ValveController(IBrokerService kafkaService)
+		{
+			_kafkaService = kafkaService;
+		}
+
+		[HttpPost("farm/{farmId}/greenhouse/{ghId}/watering")]
+		public async Task<IActionResult> interactValve([FromBody] ValveRequest request, int farmId, int ghId)
+		{
+			var kafkaRequest = new
+			{
+				FarmId = farmId,
+				GreenHouseId = ghId,
+				SomeAction = request.Action,
+			};
+
+			var message = JsonSerializer.Serialize(kafkaRequest);
+			return Ok(kafkaRequest);
+			
+			/*await _kafkaService.ProduceAsync("ValvesHeatersRequest", message);
+
+			return Ok($"Valve status is {request.Action}");*/
+		}
+	}
+}

Cloud/Controllers/WeatherForecastController.cs

@@ -1,32 +0,0 @@
-using Microsoft.AspNetCore.Mvc;
-
-namespace Cloud.Controllers;
-
-[ApiController]
-[Route("[controller]")]
-public class WeatherForecastController : ControllerBase
-{
-    private static readonly string[] Summaries = new[]
-    {
-        "Freezing", "Bracing", "Chilly", "Cool", "Mild", "Warm", "Balmy", "Hot", "Sweltering", "Scorching"
-    };
-
-    private readonly ILogger<WeatherForecastController> _logger;
-
-    public WeatherForecastController(ILogger<WeatherForecastController> logger)
-    {
-        _logger = logger;
-    }
-
-    [HttpGet(Name = "GetWeatherForecast")]
-    public IEnumerable<WeatherForecast> Get()
-    {
-        return Enumerable.Range(1, 5).Select(index => new WeatherForecast
-        {
-            Date = DateTime.Now.AddDays(index),
-            TemperatureC = Random.Shared.Next(-20, 55),
-            Summary = Summaries[Random.Shared.Next(Summaries.Length)]
-        })
-        .ToArray();
-    }
-}

Cloud/Dockerfile

@@ -0,0 +1,30 @@
+FROM mcr.microsoft.com/dotnet/aspnet:6.0 AS base
+WORKDIR /app
+EXPOSE 5124
+
+ENV ASPNETCORE_URLS=http://+:5124
+
+# Creates a non-root user with an explicit UID and adds permission to access the /app folder
+# For more info, please refer to https://aka.ms/vscode-docker-dotnet-configure-containers
+RUN adduser -u 5678 --disabled-password --gecos "" appuser && chown -R appuser /app
+USER appuser
+
+FROM mcr.microsoft.com/dotnet/sdk:6.0 AS build
+ARG configuration=Development
+WORKDIR /src
+COPY ["Cloud.csproj", "."]
+RUN dotnet restore "./Cloud.csproj"
+COPY . .
+WORKDIR "/src/."
+RUN dotnet build "./Cloud.csproj" -c $configuration -o /app/build
+
+FROM build AS publish
+ARG configuration=Development
+RUN dotnet publish "./Cloud.csproj" -c $configuration -o /app/publish /p:UseAppHost=false
+
+FROM base AS final
+WORKDIR /app
+COPY --from=publish /app/publish .
+
+ENTRYPOINT ["dotnet", "Cloud.dll"]
+

Cloud/Enums/ValveEnum.cs

@@ -0,0 +1,9 @@
+namespace Cloud.Enums
+{
+	public enum ValveEnum
+	{
+		Open,
+		Close,
+		Auto
+	}
+}

Cloud/Middlewares/DatabaseMiddleware.cs

@@ -0,0 +1,30 @@
+using Microsoft.EntityFrameworkCore;
+
+namespace Cloud.Middlewares;
+
+public static class DatabaseMiddleware
+{
+    public static void AddDbConnectionService(this IServiceCollection services)
+    {
+        string connectionString = Environment.GetEnvironmentVariable("DB_CONNECTION_STRING") 
+            ?? "Host=localhost;Port=5438;Database=main_database;Username=postgres;Password=12345";
+
+        services.AddDbContext<ApplicationContext>(options =>
+            options.UseNpgsql(connectionString));
+
+    }
+    public static void MigrateDb(this IApplicationBuilder app)
+    {
+        try 
+        {
+            using var scope = app.ApplicationServices.CreateScope();
+            var context = scope.ServiceProvider.GetRequiredService<ApplicationContext>();
+            
+            context.Database.Migrate();
+        }
+        catch (Exception ex) 
+        {
+            Console.WriteLine(ex.Message);
+        }
+    }
+}

Cloud/Models/Farm.cs

@@ -6,7 +6,7 @@
         public string Name { get; set; }
         public int UserId { get; set; }
         public User? User { get; set; }
-		public string RaspberryMacAddr { get; set; }
+        public string RaspberryIP { get; set; }
-
+        List<Greenhouse> Greenhouses { get; set; } = new();
     }
 }

Cloud/Models/Greenhouse.cs

@@ -0,0 +1,14 @@
+using Cloud.Models.Support;
+
+namespace Cloud.Models
+{
+    public class Greenhouse
+    {
+        public int Id { get; set; }
+        public int RecomendedTemperature { get; set; }
+        public WateringMode WateringMode { get; set; }
+        public HeatingMode HeatingMode { get; set; }
+        public int FarmId { get; set; }
+        public Farm? Farm { get; set; }
+    }
+}

Cloud/Models/Support/HeatingMode.cs

@@ -0,0 +1,8 @@
+namespace Cloud.Models.Support 
+{
+    public enum HeatingMode 
+    {
+        Manual,
+        Auto
+    }
+}

Cloud/Models/Support/WateringMode.cs

@@ -0,0 +1,8 @@
+namespace Cloud.Models.Support 
+{
+    public enum WateringMode 
+    {
+        Manual,
+        Auto
+    }
+}

Cloud/Program.cs

@@ -7,17 +7,30 @@ using FluentValidation;
 using FluentValidation.AspNetCore;
 using Cloud.Validation;
 using StackExchange.Redis;
+using Cloud.Services.Broker.Implement.Kafka;
+using Cloud.Services.Broker;
+using Cloud.Services;
+using Cloud.Services.Domain.Implement;
+using Cloud.Services.Domain;
+using Cloud.Services.Cache;
+using Cloud.Support;
+using System.Text.RegularExpressions;
+using Cloud.Middlewares;
 
 var builder = WebApplication.CreateBuilder(args);
 
 // Add services to the container.
+builder.Services.AddSingleton<IBrokerService, KafkaService>();
+builder.Services.AddTransient<IGreenhouseService, GreenhouseService>();
 
 //Redis configuration
+string redisUrl = Environment.GetEnvironmentVariable("REDIS_URL") ?? "localhost:6379";
 builder.Services.AddSingleton<IConnectionMultiplexer>(sp =>
 {
-    var configuration = ConfigurationOptions.Parse("localhost:6379");
+    var configuration = ConfigurationOptions.Parse(redisUrl);
     return ConnectionMultiplexer.Connect(configuration);
 });
+builder.Services.AddSingleton<IRedisCacheService, RedisCacheService>();
 
 //Jwt configuration
 var jwtIssuer = builder.Configuration.GetSection("Jwt:Issuer").Get<string>();
@@ -37,16 +50,15 @@ builder.Services.AddAuthentication(JwtBearerDefaults.AuthenticationScheme)
          IssuerSigningKey = new SymmetricSecurityKey(Encoding.UTF8.GetBytes(jwtKey))
      };
  });
-
+// Настройка подключения к БД
-builder.Services.AddDbContext<ApplicationContext>(options =>
+builder.Services.AddDbConnectionService();
-    options.UseNpgsql("Host=localhost;Port=5438;Database=main_database;Username=postgres;Password=12345"));
-
 // Настройка CORS
+string frontUrl = Environment.GetEnvironmentVariable("FRONT_URL") ?? "http://localhost:3000";
 builder.Services.AddCors(options =>
 {
     options.AddPolicy("AllowFrontendLocalhost", builder =>
     {
-        builder.WithOrigins("http://localhost:3000") // фронтенд
+        builder.WithOrigins(frontUrl) // фронтенд
                .AllowAnyHeader()
                .AllowAnyMethod();
     });
@@ -58,6 +70,7 @@ builder.Services.AddFluentValidationClientsideAdapters();
 builder.Services.AddValidatorsFromAssemblyContaining<LoginValidator>();
 builder.Services.AddValidatorsFromAssemblyContaining<RegisterValidator>();
 builder.Services.AddValidatorsFromAssemblyContaining<FarmValidator>();
+builder.Services.AddValidatorsFromAssemblyContaining<ValveValidator>();
 
 // Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
 builder.Services.AddEndpointsApiExplorer();
@@ -94,6 +107,7 @@ var app = builder.Build();
 // Configure the HTTP request pipeline.
 if (app.Environment.IsDevelopment())
 {
+    Console.WriteLine("Swagger enabled");
     app.UseSwagger();
     app.UseSwaggerUI(c =>
     {
@@ -107,6 +121,9 @@ app.UseHttpsRedirection();
 // Включение CORS
 app.UseCors("AllowFrontendLocalhost");
 
+// Применение миграций
+app.MigrateDb();
+
 app.UseAuthentication();
 
 app.UseAuthorization();

Cloud/Requests/FarmRequest.cs

@@ -3,6 +3,6 @@
     public class FarmRequest
     {
         public string Name { get; set; }
-		public string RaspberryMacAddr { get; set; }
+        public string RaspberryIP { get; set; }
     }
 }

Cloud/Requests/GreenhouseRequest.cs

@@ -0,0 +1,11 @@
+using Cloud.Models.Support;
+
+namespace Cloud.Requests
+{
+    public class GreenhouseRequest 
+    {
+        public int RecomendedTemperature { get; set; }
+        public WateringMode WateringMode { get; set; }
+        public HeatingMode HeatingMode { get; set; }
+    }
+}

Cloud/Requests/ValveRequest.cs

@@ -0,0 +1,7 @@
+namespace Cloud.Requests
+{
+	public class ValveRequest
+	{
+		public string Action { get; set; }
+	}
+}

Cloud/Services/Broker/IBrokerConsumer.cs

@@ -0,0 +1,9 @@
+using Cloud.Services.Broker.Support;
+
+namespace Cloud.Services.Broker
+{
+    public interface IBrokerConsumer 
+    {
+        IEnumerable<T>? WaitMessages<T>(string topic) where T : IBrokerResponse;
+    }
+}

Cloud/Services/Broker/IBrokerProducer.cs

@@ -0,0 +1,9 @@
+using Cloud.Services.Broker.Support;
+
+namespace Cloud.Services.Broker
+{
+    public interface IBrokerProducer 
+    {
+        Task ProduceAsync(string topic, Command command);
+    }
+}

Cloud/Services/Broker/IBrokerService.cs

@@ -0,0 +1,9 @@
+using Cloud.Services.Broker.Support;
+
+namespace Cloud.Services.Broker
+{
+    public interface IBrokerService : IBrokerProducer, IBrokerConsumer
+    {
+        void ChangeBrokerIp(string ip);
+    }
+}

Cloud/Services/Broker/Implement/Kafka/KafkaConsumer.cs

@@ -0,0 +1,93 @@
+
+using Cloud.Services.Broker.Support;
+using Confluent.Kafka;
+using System.Diagnostics;
+using System.Text.Json;
+
+namespace Cloud.Services.Broker.Implement.Kafka
+{
+    public class KafkaConsumer : IBrokerConsumer
+    {
+        private IConsumer<string, string> _consumer;
+        private readonly IConfiguration _config;
+
+        public KafkaConsumer(IConfiguration config)
+        {
+            _config = config;
+            Console.WriteLine($"KafkaConsumer created. IP:" + _config["KAFKA_URL"]);
+            ChangeBrokerIp(_config["KAFKA_URL"]);
+        }
+
+        public IEnumerable<T>? WaitMessages<T>(string topic)
+            where T : IBrokerResponse
+        {   
+            List<T> res = new();
+            List<PartitionMetadata> partitions;
+
+            using var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = _config["KAFKA_URL"] }).Build();
+            var meta = adminClient.GetMetadata(TimeSpan.FromSeconds(20));
+            var currentTopic = meta.Topics.SingleOrDefault(t => t.Topic == topic) 
+                ?? throw new Exception($"Topic {topic} not found");
+            partitions = currentTopic.Partitions;
+            
+            _consumer.Subscribe(topic);
+            foreach (var partition in partitions)
+            {
+                var topicPartition = new TopicPartition(topic, partition.PartitionId);
+                _consumer.Assign(topicPartition);
+
+                T? message = _consume<T>();
+                if (message == null) return null;
+                res.Add(message);
+                
+            }
+            _consumer.Unassign();
+            _consumer.Unsubscribe();
+            return res;
+        }
+
+        private T? _consume<T>() where T : IBrokerResponse
+        {
+            var sw = new Stopwatch();
+            sw.Start();
+            try
+            {
+                while (true)
+                {
+                    var consumeResult = _consumer.Consume(TimeSpan.FromMinutes(1));
+
+                    if (consumeResult?.Message?.Value == null)
+                    {
+                        // Предел по времени
+                        if (sw.Elapsed > TimeSpan.FromMinutes(1))
+                        {
+                            return default;
+                        }
+                        continue;
+                    }
+
+                    string jsonObj = consumeResult.Message.Value;
+                    return JsonSerializer.Deserialize<T>(jsonObj);
+
+                }
+            }
+            catch (Exception ex)
+            {
+                _consumer.Close();
+                throw;
+            }
+        }
+
+        public void ChangeBrokerIp(string ip)
+        {
+            var consumerConfig = new ConsumerConfig() 
+            {
+                BootstrapServers = ip,
+				GroupId = _config["Kafka:GroupId"],
+				AutoOffsetReset = AutoOffsetReset.Earliest,
+            };
+            _consumer?.Close();
+            _consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
+        }
+    }
+}

Cloud/Services/Broker/Implement/Kafka/KafkaProducer.cs

@@ -0,0 +1,38 @@
+using System.Text.Json;
+using Cloud.Services.Broker.Support;
+using Confluent.Kafka;
+
+namespace Cloud.Services.Broker.Implement.Kafka
+{
+    public class KafkaProducer : IBrokerProducer
+    {
+        private IProducer<string, string> _producer;
+		private readonly IConfiguration _config;
+
+		public KafkaProducer(IConfiguration configuration)
+		{
+			_config = configuration;
+			Console.WriteLine($"KafkaProducer created. IP:" + _config["KAFKA_URL"]);
+            ChangeBrokerIp(_config["KAFKA_URL"]);
+		}
+		public async Task ProduceAsync(string topic, Command command)
+		{
+			var commandSerialized = JsonSerializer.Serialize(command);
+            var message = new Message<string, string> { Key = Guid.NewGuid().ToString(), Value = commandSerialized };
+            
+			//Produce the Message
+			await _producer.ProduceAsync(topic, message);
+		}
+
+        public void ChangeBrokerIp(string ip)
+        {
+			var producerConfig = new ProducerConfig
+			{
+				BootstrapServers = ip
+			};
+
+			//Build the Producer
+			_producer = new ProducerBuilder<string, string>(producerConfig).Build();
+        }
+    }
+}

Cloud/Services/Broker/Implement/Kafka/KafkaService.cs

@@ -0,0 +1,30 @@
+using Cloud.Services.Broker.Support;
+
+namespace Cloud.Services.Broker.Implement.Kafka
+{
+    public class KafkaService : IBrokerService
+    {
+        private readonly KafkaProducer _producer;
+        private readonly KafkaConsumer _consumer;
+
+        public KafkaService(IConfiguration configuration) 
+        {
+            _producer = new KafkaProducer(configuration);
+            _consumer = new KafkaConsumer(configuration);
+        }
+
+        public IEnumerable<T>? WaitMessages<T>(string topic) 
+            where T : IBrokerResponse
+            => _consumer.WaitMessages<T>(topic);
+
+
+        public async Task ProduceAsync(string topic, Command command)
+            => await _producer.ProduceAsync("commands", command);
+
+        public void ChangeBrokerIp(string ip)
+        {
+            _consumer.ChangeBrokerIp(ip);
+            _producer.ChangeBrokerIp(ip);
+        }
+    }
+}

Cloud/Services/Broker/Support/Command.cs

@@ -0,0 +1,10 @@
+using System.Text.Json;
+
+namespace Cloud.Services.Broker.Support
+{
+    public class Command
+    {
+        public Guid GreenhouseId { get; set; }
+        public string CommandName { get; set; } = null!;
+    }
+}

Cloud/Services/Broker/Support/CommandResult.cs

@@ -0,0 +1,9 @@
+namespace Cloud.Services.Broker.Support
+{
+    public class CommandResult : IBrokerResponse
+    {
+        public int CommandId { get; set; }
+        public int GreenhouseId { get; set; }
+        public string ResultMessage { get; set; } = string.Empty;
+    }
+}

Cloud/Services/Broker/Support/GreenhouseInfo.cs

@@ -0,0 +1,12 @@
+namespace Cloud.Services.Broker.Support
+{
+    public class GreenhouseInfo : IBrokerResponse
+    {
+        public int Id { get; set; }
+        public int PercentWater { get; set; }
+        public int SoilTemperature { get; set; }
+        public bool PumpStatus { get; set; }
+        public bool HeatingStatus { get; set; }
+        public bool AutoWateringStatus { get; set; }
+    }
+}

Cloud/Services/Broker/Support/IBrokerResponse.cs

@@ -0,0 +1,6 @@
+namespace Cloud.Services.Broker.Support
+{
+    public interface IBrokerResponse 
+    {
+    }
+}

Cloud/Services/Cache/IRedisCacheService.cs

@@ -0,0 +1,13 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace Cloud.Services.Cache
+{
+    public interface IRedisCacheService
+    {
+        Task SetCacheAsync<T>(string key, T value, TimeSpan? expiry = null);
+        Task<T?> GetCacheAsync<T>(string key);
+    }
+}

Cloud/Services/Cache/RedisCacheService.cs

@@ -0,0 +1,37 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using StackExchange.Redis;
+using System.Text.Json;
+
+namespace Cloud.Services.Cache
+{
+    public class RedisCacheService : IRedisCacheService
+    {
+        private readonly IConnectionMultiplexer _connectionMultiplexer;
+
+        public RedisCacheService(IConnectionMultiplexer connectionMultiplexer)
+        {
+            _connectionMultiplexer = connectionMultiplexer;
+        }
+
+        public async Task SetCacheAsync<T>(string key, T value, TimeSpan? expiry = null)
+        {
+            var database = _connectionMultiplexer.GetDatabase();
+            var serializedValue = JsonSerializer.Serialize(value);
+            await database.StringSetAsync(key, serializedValue, expiry);
+        }
+
+        public async Task<T?> GetCacheAsync<T>(string key)
+        {
+            var database = _connectionMultiplexer.GetDatabase();
+            var value = await database.StringGetAsync(key);
+
+            if (value.IsNullOrEmpty)
+                return default;
+
+            return JsonSerializer.Deserialize<T>(value);
+        }
+    }
+}

Cloud/Services/Domain/IGreenhouseService.cs

@@ -0,0 +1,45 @@
+using Cloud.Models;
+using Cloud.Services.Broker.Support;
+
+namespace Cloud.Services.Domain;
+
+public interface IGreenhouseService
+{
+    /// <summary>
+    /// Возвращает текущую информацию о конкретной теплице из брокера
+    /// </summary>
+    /// <param name="id">ID теплицы</param>
+    /// <param name="farmId">ID фермы, то есть брокера</param>
+    /// <returns>Текущие данные о теплице от менеджера теплицы</returns>
+    public Task<GreenhouseInfo?> GetGreenhouseInfo(int id, int farmId);
+    /// <summary>
+    /// Возвращает сохраненные данные для автоматизации теплицы из базы данных
+    /// </summary>
+    /// <param name="id">ID теплицы</param>
+    /// <returns>Данные для автоматизации теплицы</returns>
+    public Task<Greenhouse?> GetGreenhouse(int id);
+    /// <summary>
+    /// Возвращает список данных о всех теплицах пользователя из брокера
+    /// </summary>
+    /// <param name="farmId">ID фермы</param>
+    /// <returns>Список текущих данных о теплицах</returns>
+    public Task<IEnumerable<GreenhouseInfo>?> GetAll(int farmId);
+    /// <summary>
+    /// Сохраняет данные об автоматизации теплицы в базу данных
+    /// </summary>
+    /// <param name="greenhouse">Данные автоматизации теплицы</param>
+    /// <returns>Созданную сущность из базы данных</returns>
+    public Task<Greenhouse> Create(Greenhouse greenhouse);
+    /// <summary>
+    /// Обновляет данные автоматизации теплицы в базе данных
+    /// </summary>
+    /// <param name="greenhouse">Новая информация об автоматизации теплицы</param>
+    /// <returns>Обновленную сущность из базы данных</returns>
+    public Task<Greenhouse> Update(Greenhouse greenhouse);
+    /// <summary>
+    /// Удаляет данные об автоматизации теплицы из базы данных
+    /// </summary>
+    /// <param name="id">ID данных автоматизации теплицы</param>
+    /// <returns>Возвращает удаленную сущность</returns>
+    public Task<Greenhouse> Delete(int id);
+}

Cloud/Services/Domain/Implement/GreenhouseService.cs

@@ -0,0 +1,67 @@
+using Cloud.Models;
+using Cloud.Services.Broker;
+using Cloud.Services.Broker.Support;
+using Microsoft.EntityFrameworkCore;
+
+namespace Cloud.Services.Domain.Implement;
+
+public class GreenhouseService : IGreenhouseService
+{
+    private readonly IBrokerService _brokerService;
+    private readonly ApplicationContext _context;
+
+    public GreenhouseService(IBrokerService brokerService, ApplicationContext context)
+    {
+        _context = context;
+        _brokerService = brokerService;
+    }
+
+    public async Task<Greenhouse> Create(Greenhouse greenhouse)
+    {
+        var res = await _context.Greenhouses.AddAsync(greenhouse);
+        await _context.SaveChangesAsync();
+        return res.Entity;
+    }
+
+    public async Task<Greenhouse> Delete(int id)
+    {
+        var greenhouse = await _context.Greenhouses.FirstOrDefaultAsync(x => x.Id == id);
+
+        _context.Greenhouses.Remove(greenhouse);
+        await _context.SaveChangesAsync();
+        return greenhouse;
+    }
+
+
+    public async Task<Greenhouse?> GetGreenhouse(int id)
+    {
+        return await _context.Greenhouses.FirstOrDefaultAsync(x => x.Id == id);
+    }
+
+    public async Task<Greenhouse> Update(Greenhouse greenhouse)
+    {
+        var res = _context.Greenhouses.Update(greenhouse);
+        await _context.SaveChangesAsync();
+        return res.Entity;
+    }
+
+    public async Task<IEnumerable<GreenhouseInfo>?> GetAll(int farmId)
+    {
+        // await _changeBrokerIp(farmId);
+        return _brokerService.WaitMessages<GreenhouseInfo>("data");
+    }
+    
+    public async Task<GreenhouseInfo?> GetGreenhouseInfo(int id, int farmId)
+    {
+        // await _changeBrokerIp(farmId);
+        var infos = _brokerService.WaitMessages<GreenhouseInfo>("data");
+        return infos?.FirstOrDefault(x => x.Id == id);
+    }
+
+    private async Task _changeBrokerIp(int farmId)
+    {
+        var farm = await _context.Farms.FirstOrDefaultAsync(x => x.Id == farmId);
+        _brokerService.ChangeBrokerIp(farm.RaspberryIP);
+
+    }
+}

Cloud/Support/NetworkSupport.cs

@@ -0,0 +1,26 @@
+namespace Cloud.Support;
+
+public static class NetworkSupport
+{
+    public static async Task CheckConnectionAsync(string address)
+    {
+        using var client = new HttpClient();
+        try
+        {
+            var response = await client.GetAsync(address);
+
+            if (response.IsSuccessStatusCode)
+            {
+                Console.WriteLine($"Соединение успешно проверено. Статус-код: {response.StatusCode}");
+            }
+            else
+            {
+                Console.WriteLine($"Соединение не удалось проверить. Статус-код: {response.StatusCode}. URL: {address}");
+            }
+        }
+        catch (HttpRequestException ex)
+        {
+            Console.WriteLine($"Ошибка при проверке соединения: {ex.Message}. URL: {address}");
+        }
+    }
+}

Cloud/Validation/FarmValidator.cs

@@ -7,9 +7,9 @@ namespace Cloud.Validation
     {
         public FarmValidator()
         {
-			RuleFor(request => request.RaspberryMacAddr)
+            RuleFor(request => request.RaspberryIP)
-				.NotEmpty().WithMessage("MAC address can't be empty")
+                .NotEmpty().WithMessage("IP address can't be empty")
-				.Matches("^([0-9A-Fa-f]{2}[:-]?){5}([0-9A-Fa-f]{2})$").WithMessage("MAC address is not valid");
+                .Matches(@"^((25[0-5]|(2[0-4]|1\d|[1-9]|)\d)\.?\b){4}$").WithMessage("IP address is not valid");
 
             RuleFor(request => request.Name)
                 .NotEmpty().WithMessage("Name can't be empty");

Cloud/Validation/ValveValidator.cs

@@ -0,0 +1,16 @@
+using Cloud.Enums;
+using Cloud.Requests;
+using FluentValidation;
+
+namespace Cloud.Validation
+{
+	public class ValveValidator : AbstractValidator<ValveRequest>
+	{
+		public ValveValidator() {
+
+			RuleFor(request => request.Action)
+				.NotEmpty().WithMessage("Action can't be empty").
+				IsEnumName(typeof (ValveEnum)).WithMessage("Action is not correct");
+		}
+	}
+}

Cloud/WeatherForecast.cs

@@ -1,12 +0,0 @@
-namespace Cloud;
-
-public class WeatherForecast
-{
-    public DateTime Date { get; set; }
-
-    public int TemperatureC { get; set; }
-
-    public int TemperatureF => 32 + (int)(TemperatureC / 0.5556);
-
-    public string? Summary { get; set; }
-}

Cloud/appsettings.json

@@ -5,6 +5,10 @@
       "Microsoft.AspNetCore": "Warning"
     }
   },
+  "Kafka": {
+    "BootstrapServers": "localhost:9092",
+    "GroupId": "ValvesHeaters"
+  },
   "AllowedHosts": "*",
   "Jwt": {
     "Key": "m7TyhE20s0dVtUDAr9EnFdPZnAG8maxgBTaiW5j6kO6RQhWDAGxYmXyu0suDnE0o",

GreenhouseController/ManageController.py

@@ -0,0 +1,48 @@
+from json import dumps
+
+
+
+from json import dumps
+
+class ManageController:
+    def __init__(self, producer, topic='commands'):
+        self.valve_state = "closed"  
+        self.heater_state = "off"   
+        self.producer = producer
+        self.topic = topic
+
+    def toggle_device(self, device, request_id, greenhouse_id):
+        
+        
+        if device == 'valve':
+           
+            if self.valve_state == 'closed':
+                self.valve_state = 'open'
+                print("Valve opened")
+            else:
+                self.valve_state = 'closed'
+                print("Valve closed")
+        
+        elif device == 'heater':
+          
+            if self.heater_state == 'off':
+                self.heater_state = 'on'
+                print("Heater turned on")
+            else:
+                self.heater_state = 'off'
+                print("Heater turned off")
+
+      
+        self.send_status(request_id, greenhouse_id)
+
+    def send_status(self, request_id, greenhouse_id):
+       
+        status = {
+            'request_id': request_id,
+            'greenhouse_id': greenhouse_id,
+            'valve_state': self.valve_state,
+            'heater_state': self.heater_state
+        }
+
+        print(f"Sent device status: {status}")
+        return  status

GreenhouseDetector/Dockerfile

@@ -0,0 +1,11 @@
+FROM python:3.9-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY GreenhouseDetector/detector.py .
+
+CMD ["python", "detector.py"]

GreenhouseDetector/detector.py

@@ -0,0 +1,43 @@
+import os
+import time
+import random as rnd
+
+from flask import Flask
+import requests
+import threading
+
+
+app = Flask(__name__)
+
+class Detector:
+    def __init__(self, id, moistureThresholdUpper, moistureThresholdLower, tempThresholdUpper, tempThresholdLower):
+        self.MANAGER_URL = os.environ.get('MANAGER_URL')
+        print("MANAGER_URL=", self.MANAGER_URL)
+        self.id = id
+        self.moistureThresholdUpper = moistureThresholdUpper
+        self.moistureThresholdLower = moistureThresholdLower
+        self.tempThresholdUpper = tempThresholdUpper
+        self.tempThresholdLower = tempThresholdLower
+        self.moisture = 0
+        self.temp = 0
+
+    def cycle(self):
+        self.moisture += rnd.random() / 100
+        self.temp += (rnd.random() - 0.5) / 100
+
+    def sendData(self):
+        data = {"moisture": self.moisture,
+                "temp": self.temp}
+        requests.post(f"{self.MANAGER_URL}/webhook?id={self.id}", json=data)
+
+detector1 = Detector(1, 0.6, 0.2, 40, 20)
+
+detectors = [detector1]
+
+if __name__ =="__main__":
+    while True:
+        for detector in detectors:
+            detector.cycle()
+            detector.sendData()
+        time.sleep(1)
+

GreenhouseManager/Dockerfile

@@ -0,0 +1,11 @@
+FROM python:3.9-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY GreenhouseManager/manager.py .
+
+CMD ["python", "manager.py"]

GreenhouseManager/manager.py

@@ -0,0 +1,140 @@
+import os
+from kafka import KafkaProducer, KafkaConsumer
+import kafka
+import socket
+from json import dumps, loads
+from flask import Flask, request
+import time
+from enum import Enum
+import threading
+
+
+app = Flask(__name__)
+
+def start_manager():
+    return
+
+class Manager:
+    def __init__(self, _id: int, moisture: float = 0, temp: float = 20,  isAutoOn: bool = False, valve_state: str = "closed",
+                 heater_state: str = "off"):
+        KAFKA_URL = os.environ.get('KAFKA_URL')
+        print("KAFKA_URL=", KAFKA_URL)
+        self._id = _id
+        self.moisture = moisture
+        self.temp = temp
+        self.isAutoOn = isAutoOn
+        self.valve_state = valve_state
+        self.heater_state = heater_state
+
+        self.dataPublisher = KafkaProducer(
+            bootstrap_servers=[KAFKA_URL],
+            client_id=f'manager{self._id}_producer',
+            value_serializer=lambda v: dumps(v).encode('utf-8')
+        )
+
+        self.controllerConsumer = KafkaConsumer(
+            'commands',
+            bootstrap_servers=[KAFKA_URL],
+            auto_offset_reset='earliest',
+            enable_auto_commit=True,
+            consumer_timeout_ms=2000,
+            group_id=f'manager{self._id}',
+            value_deserializer=lambda x: loads(x.decode('utf-8'))
+        )
+        self.controllerConsumerResponse = KafkaProducer(
+            bootstrap_servers=[KAFKA_URL],
+            client_id=f'manager{self._id}_producer',
+            value_serializer=lambda v: dumps(v).encode('utf-8')
+        )
+
+    def sendData(self):
+        print("sending data...")
+        message = {
+            'id': self._id,
+            'moisture': self.moisture,
+            'temp': self.temp,
+            'valveStatus': str(self.valve_state),
+            'heaterStatus': str(self.heater_state),
+            'isAutoOn': self.isAutoOn
+        }
+
+        print(message)
+        self.dataPublisher.send('data', message)
+        self.dataPublisher.flush()
+
+
+    def toggle_device(self, device, request_id, greenhouse_id):
+
+        if device == 'valve':
+
+            if self.valve_state == 'closed':
+                self.valve_state = 'open'
+                print("Valve opened")
+            else:
+                self.valve_state = 'closed'
+                print("Valve closed")
+
+        elif device == 'heater':
+
+            if self.heater_state == 'off':
+                self.heater_state = 'on'
+                print("Heater turned on")
+            else:
+                self.heater_state = 'off'
+                print("Heater turned off")
+
+        self.send_status(request_id, greenhouse_id)
+
+    def send_status(self, request_id, greenhouse_id):
+
+        status = {
+            'request_id': request_id,
+            'greenhouse_id': greenhouse_id,
+            'valve_state': self.valve_state,
+            'heater_state': self.heater_state
+        }
+        self.sendDataCommand(status)
+        print("Updating info...\n")
+
+    def sendDataCommand(self, message):
+        print("sending data...")
+
+        self.dataPublisher.send('response', message)
+
+    def getCommand(self):
+        messages = self.controllerConsumer.poll(timeout_ms=1000)
+
+        # Проверяем, есть ли сообщения
+
+        for tp, msgs in messages.items():
+            for message in msgs:
+                print(f"Manager {self._id} received message: ")
+                print(message.value)
+                self.request_id = message.value['request_id']
+                self.greenhouse_id = message.value['greenhouse_id']
+                self.command = message.value['command']
+                self.toggle_device(self.command, self.request_id, self.greenhouse_id)
+
+@app.route(f'/webhook', methods=['POST'])
+def webhook():
+    print("received webhook", request.args.get('id'))
+    for manager in managers:
+        print()
+        if int(request.args.get('id')) == manager._id and request.method == 'POST':
+            print("Data received from Webhook is", request.json)
+
+            body = request.json
+            for key, value in body.items():
+                setattr(manager, key, value)
+
+            manager.sendData()
+
+            return f"Webhook received for manager {manager._id}"
+    return "Webhook ignored"
+
+t1 = threading.Thread(target=start_manager)
+manager1 = Manager(_id=1)
+managers = [manager1]
+
+if __name__ == "__main__":
+    threading.Thread(target=lambda: app.run(host="0.0.0.0", port=20002, debug=True, use_reloader=False)).start()

docker-compose.yml

@@ -1,16 +1,40 @@
+networks:
+  vpn:
+    name: kafkaVPN
+    driver: bridge
+    ipam:
+      config:
+        - subnet: "192.168.2.0/24"
+          gateway: "192.168.2.1"
+
 services:
+  cloud:
+    networks:
+      - vpn
+    build: ./Cloud/
+    ports:
+      - "5124:5124"
+    environment:
+      ASPNETCORE_ENVIRONMENT: Development
+      DB_CONNECTION_STRING: ${DB_CONNECTION_STRING}
+      REDDIS_URL: redis:6379
+      KAFKA_URL: kafka:29092
+      # Добавить, когда будет фронт!
+      # FRONT_URL: front:3000
+    depends_on:
+      - postgres
+      - redis
   postgres:
     image: postgres:14
     container_name: cucumber_database
     environment:
-      POSTGRES_USER: postgres 
+      POSTGRES_USER: ${POSTGRES_USER}
-      POSTGRES_PASSWORD: 12345  
+      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
-      POSTGRES_DB: main_database  
+      POSTGRES_DB: ${POSTGRES_DB}
     ports:
       - "5438:5432"
     volumes:
       - postgres_data:/var/lib/postgresql/data
-  
   redis:
     image: 'redis:latest'
     ports:
@@ -24,6 +48,94 @@ services:
         - ping
       retries: 3
       timeout: 5s
+  zookeeper:
+    networks:
+      - vpn
+    image: confluentinc/cp-zookeeper:7.4.0
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    ports:
+      - 2181:2181
+  kafka:
+    networks:
+      vpn:
+        ipv4_address: 192.168.2.10
+    image: confluentinc/cp-kafka:7.4.0
+    ports:
+      - 9092:9092
+      - 9997:9997
+    expose:
+      - 29092:29092
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_LISTENERS: HOST://0.0.0.0:9092,DOCKER://0.0.0.0:29092
+      KAFKA_ADVERTISED_LISTENERS: HOST://192.168.1.5:9092,DOCKER://kafka:29092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: DOCKER:PLAINTEXT,HOST:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: DOCKER
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_LOG_FLUSH_INTERVAL_MESSAGES: 10000
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+    depends_on:
+      - zookeeper
+  init-kafka:
+    networks:
+      - vpn
+    image: confluentinc/cp-kafka:7.4.0
+    depends_on:
+      - kafka
+    entrypoint: [ '/bin/sh', '-c' ]
+    command: |
+      "
+      # blocks until kafka is reachable
+      kafka-topics --bootstrap-server kafka:29092 --list
+
+      echo -e 'Creating kafka topics'
+      kafka-topics --bootstrap-server kafka:29092 --create --if-not-exists --topic commands --replication-factor 1 --partitions 1
+      kafka-topics --bootstrap-server kafka:29092 --create --if-not-exists --topic data --replication-factor 1 --partitions 1
+      kafka-topics --bootstrap-server kafka:29092 --create --if-not-exists --topic response --replication-factor 1 --partitions 1
+
+      echo -e 'Successfully created the following topics:'
+      kafka-topics --bootstrap-server kafka:29092 --list
+      "
+  kafka-ui:
+    networks:
+      - vpn
+    container_name: kafka-ui
+    image: provectuslabs/kafka-ui:latest
+    ports:
+      - 8080:8080
+    depends_on:
+      - kafka
+    environment:
+      KAFKA_CLUSTERS_0_NAME: local
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
+      KAFKA_CLUSTERS_0_METRICS_PORT: 9997
+  manager:
+    networks:
+      - vpn
+    build:
+      context: .
+      dockerfile: ./GreenhouseManager/Dockerfile
+    environment:
+      KAFKA_URL: kafka:29092
+    depends_on:
+      - kafka
+    expose:
+      - 20002
+  detector:
+    networks:
+      - vpn
+    build:
+      context: .
+      dockerfile: ./GreenhouseDetector/Dockerfile
+    environment:
+      MANAGER_URL: http://manager:20002
+    depends_on:
+      - manager
+
 volumes:
   postgres_data:
     driver: local

requirements.txt

@@ -0,0 +1,3 @@
+kafka-python~=2.0.2
+Flask~=3.0.3
+requests~=2.31.0