Ticket #62 : Add bridge between two servers

Author: habarthierry-hue

.gitignore

@@ -212,4 +212,7 @@ log-*.txt
 docs/_build/
 
 ## Ignore Javascript
-wwwroot/
+wwwroot/
+
+# Ignore DB
+*.db

src/EventMesh.Runtime.Website/Program.cs

@@ -5,7 +5,6 @@
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using System;
-using System.Collections.Generic;
 using System.IO;
 using System.Reflection;
 
@@ -19,6 +18,7 @@ public static void Main(string[] args)
             // LaunchAMQPEventMeshServer();
             // LaunchKafkaEventMeshServer();
             LaunchFullEventMeshServer();
+            LaunchFullEventMeshServer(4001, "Runtime2.db");
             CreateHostBuilder(args).Build().Run();
         }
 
@@ -87,15 +87,15 @@ private static IRuntimeHost LaunchKafkaEventMeshServer()
             return runtimeHost;
         }
 
-        private static IRuntimeHost LaunchFullEventMeshServer()
+        private static IRuntimeHost LaunchFullEventMeshServer(int port = 4000, string dbName = "Runtime.db")
         {
             var migrationsAssembly = typeof(Program).GetTypeInfo().Assembly.GetName().Name;
             Console.WriteLine("Launch EventMesh server...");
-            var path = Path.Combine(Environment.CurrentDirectory, "Runtime.db");
+            var path = Path.Combine(Environment.CurrentDirectory, dbName);
             var builder = new RuntimeHostBuilder(opt =>
             {
                 opt.Urn = "localhost";
-                opt.Port = 4000;
+                opt.Port = port;
             })
                 .AddKafka()
                 .AddAMQP()

src/EventMesh.Runtime.Website/Runtime.db

@@ -1,13 +1,16 @@
-using System;
+using FaasNet.Gateway.Core.Exceptions;
+using FaasNet.Gateway.Core.Resources;
+using System;
 using System.Collections.Generic;
+using System.Linq;
 
 namespace FaasNet.Gateway.Core.Domains
 {
     public class EventMeshServerAggregate
     {
         public EventMeshServerAggregate()
         {
-            Bridges = new List<EventMeshServerAggregate>();
+            Bridges = new List<EventMeshServerBridge>();
         }
 
         public string Urn { get; set; }
@@ -17,7 +20,7 @@ public EventMeshServerAggregate()
         public double? Latitude { get; set; }
         public double? Longitude { get; set; }
         public DateTime CreateDateTime { get; set; }
-        public ICollection<EventMeshServerAggregate> Bridges { get; set; }
+        public ICollection<EventMeshServerBridge> Bridges { get; set; }
 
         public static EventMeshServerAggregate Create(string urn, int port, string countryIsoCode, string postalCode, double? latitude, double? longitude)
         {
@@ -32,5 +35,19 @@ public static EventMeshServerAggregate Create(string urn, int port, string count
                 Urn = urn 
             };
         }
+
+        public void AddBridge(string urn, int port)
+        {
+            if(Bridges.Any(b => b.Urn == urn && b.Port == port))
+            {
+                throw new DomainException(ErrorCodes.BridgeAlreadyExists, Global.BridgeAlreadyExists);
+            }
+
+            Bridges.Add(new EventMeshServerBridge
+            {
+                Port = port,
+                Urn = urn
+            });
+        }
     }
 }

src/FaasNet.Gateway.Core/Domains/EventMeshServerAggregate.cs

@@ -0,0 +1,8 @@
+namespace FaasNet.Gateway.Core.Domains
+{
+    public class EventMeshServerBridge
+    {
+        public string Urn { get; set; }
+        public int Port { get; set; }
+    }
+}

src/FaasNet.Gateway.Core/Domains/EventMeshServerBridge.cs

@@ -10,8 +10,10 @@ public static class ErrorCodes
         public const string UnknownStateMachine = "UnknownStateMachine";
         public const string UnknownOpenApiOperation = "UnknownOpenApiOperation";
         public const string UnknownAsyncApiOperation = "UnknownAsyncApiOperation";
+        public const string UnknownEventMeshServer = "UnknownEventMeshServer";
         public const string StateMachineExists = "StateMachineExists";
         public const string FunctionExists = "FunctionExists";
         public const string EventMeshServerExists = "EventMeshServerExists";
+        public const string BridgeAlreadyExists = "BridgeAlreadyExists";
     }
 }

src/FaasNet.Gateway.Core/ErrorCodes.cs

@@ -0,0 +1,16 @@
+using MediatR;
+
+namespace FaasNet.Gateway.Core.EventMeshServer.Commands
+{
+    public class AddEventMeshServerBridgeCommand : IRequest<bool>
+    {
+        public EventMeshServer From { get; set; }
+        public EventMeshServer To { get; set; }
+    }
+
+    public class EventMeshServer
+    {
+        public string Urn { get; set; }
+        public int Port { get; set; }
+    }
+}

src/FaasNet.Gateway.Core/EventMeshServer/Commands/AddEventMeshServerBridgeCommand.cs

@@ -0,0 +1,58 @@
+using EventMesh.Runtime;
+using EventMesh.Runtime.Exceptions;
+using FaasNet.Gateway.Core.Exceptions;
+using FaasNet.Gateway.Core.Repositories;
+using FaasNet.Gateway.Core.Resources;
+using MediatR;
+using Microsoft.Extensions.Logging;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace FaasNet.Gateway.Core.EventMeshServer.Commands.Handlers
+{
+    public class AddEventMeshServerBridgeCommandHandler : IRequestHandler<AddEventMeshServerBridgeCommand, bool>
+    {
+        private readonly IEventMeshServerRepository _eventMeshServerRepository;
+        private readonly ILogger<AddEventMeshServerBridgeCommandHandler> _logger;
+
+        public AddEventMeshServerBridgeCommandHandler(IEventMeshServerRepository eventMeshServerRepository, ILogger<AddEventMeshServerBridgeCommandHandler> logger)
+        {
+            _eventMeshServerRepository = eventMeshServerRepository;
+            _logger = logger;
+        }
+
+        public async Task<bool> Handle(AddEventMeshServerBridgeCommand request, CancellationToken cancellationToken)
+        {
+            var fromServer = await _eventMeshServerRepository.Get(request.From.Urn, request.From.Port);
+            if(fromServer == null)
+            {
+                throw new BadRequestException(ErrorCodes.UnknownEventMeshServer, string.Format(Global.UnknownEventMeshServer, request.From.Urn, request.From.Port));
+            }
+
+            var toServer = await _eventMeshServerRepository.Get(request.To.Urn, request.To.Port);
+            if(toServer == null)
+            {
+                throw new BadRequestException(ErrorCodes.UnknownEventMeshServer, string.Format(Global.UnknownEventMeshServer, request.To.Urn, request.To.Port));
+            }
+
+            fromServer.AddBridge(request.To.Urn, request.To.Port);
+            await AddBridge(request.From, request.To, cancellationToken);
+            await _eventMeshServerRepository.Update(fromServer, cancellationToken);
+            await _eventMeshServerRepository.SaveChanges(cancellationToken);
+            return true;
+        }
+        private async Task AddBridge(EventMeshServer from, EventMeshServer to, CancellationToken cancellationToken)
+        {
+            try
+            {
+                var runtimeClient = new RuntimeClient(from.Urn, from.Port);
+                await runtimeClient.AddBridge(to.Urn, to.Port, cancellationToken);
+            }
+            catch (RuntimeClientException ex)
+            {
+                _logger.LogError(ex.ToString());
+                throw new EventMeshSeverUnreachableException(ex.Message);
+            }
+        }
+    }
+}

src/FaasNet.Gateway.Core/EventMeshServer/Commands/Handlers/AddEventMeshServerBridgeCommandHandler.cs

@@ -34,13 +34,12 @@ public async Task<EventMeshServerResult> Handle(AddEventMeshServerCommand reques
         {
             var urn = request.IsLocalhost ? Constants.DefaultUrl : request.Urn;
             var port = request.Port ?? Constants.DefaultPort;
-            var eventMeshServer = _eventMeshServerRepository.Get(urn, port);
-            if (eventMeshServer == null)
+            var eventMeshServer = await _eventMeshServerRepository.Get(urn, port);
+            if (eventMeshServer != null)
             {
                 throw new BadRequestException(ErrorCodes.EventMeshServerExists, string.Format(Global.EventMeshServerExists, urn, port));
             }
 
-
             await CheckEventMeshServer(urn, port, cancellationToken);
             var city = await GetCity(request.IsLocalhost, request.Urn);
             var record = EventMeshServerAggregate.Create(urn, port, city.Country.IsoCode, city.Postal.Code, city.Location.Latitude, city.Location.Longitude);
@@ -65,7 +64,7 @@ private async Task CheckEventMeshServer(string urn, int port, CancellationToken
 
         private async Task<CityResponse> GetCity(bool isLocalhost, string urn)
         {
-            CityResponse cityResponse = null;
+            CityResponse cityResponse = null; // new CityResponse(country: new MaxMind.GeoIP2.Model.Country(isoCode: "BE"), postal: new MaxMind.GeoIP2.Model.Postal(code: "1342"), location: new MaxMind.GeoIP2.Model.Location(longitude: 4.555060, latitude: 50.676090));
             if (isLocalhost)
             {
                 cityResponse = await _maxMindClient.CityAsync();

src/FaasNet.Gateway.Core/EventMeshServer/Commands/Handlers/AddEventMeshServerCommandHandler.cs

@@ -0,0 +1,19 @@
+using FaasNet.Gateway.Core.Domains;
+
+namespace FaasNet.Gateway.Core.EventMeshServer.Queries.Results
+{
+    public class EventMeshServerBridgeResult
+    {
+        public string Urn { get; set; }
+        public int Port { get; set; }
+
+        public static EventMeshServerBridgeResult ToDto(EventMeshServerBridge parameter)
+        {
+            return new EventMeshServerBridgeResult
+            {
+                Urn = parameter.Urn,
+                Port = parameter.Port
+            };
+        }
+    }
+}