Merge pull request #104 from buildersoftdev/v2.1/main

V2.1/main

README.md

@@ -1,12 +1,12 @@
-What is Andy X Data Storage?
+What is Andy X Storage?
 ============
 
-Andy X Data Storage is an open-source standalone service that is used to store messages for Andy X. The Data Storage service is offers support for Multitenancy storage. X Data Storage hosts all messages and makes sure that all of them are readable for the client.
+Andy X Storage is an open-source standalone service that is used to store messages for Andy X. The Data Storage service is offers support for Multitenancy storage. X Data Storage hosts all messages and makes sure that all of them are readable for the client.
 
 
 ## Get Started
 
-Follow the [Getting Started](https://andyx.azurewebsites.net/) instructions how to run Andy X.
+Follow the [Getting Started](https://buildersoftdev.azurewebsites.net/andyx) instructions how to run Andy X.
 
 For local development and testing, you can run Andy X within a Docker container, for more info click [here](https://hub.docker.com/u/buildersoftdev)
 
@@ -23,8 +23,8 @@ Security issues and bugs should be reported privately, via email, en.buildersoft
 
 These are some other repos for related projects:
 
-* [Andy X Dashboard](https://github.com/buildersoftdev/andyxdashboard) - Dashboard for Andy X Node
-* [Andy X Terminal](https://github.com/buildersoftdev/andyxterminal) - Manage all resources of Andy X
+* [Andy X](https://github.com/buildersoftdev/andyx) - Andy X Node, consume and produce messages
+* [Andy X Cli](https://github.com/buildersoftdev/andyx-cli) - Manage all resources of Andy X
 
 ## Deploying Andy X Data Storage with docker-compose
 

src/Storage.App/Dockerfile

@@ -2,8 +2,8 @@
 
 FROM mcr.microsoft.com/dotnet/aspnet:6.0 AS base
 WORKDIR /app
-EXPOSE 80
-EXPOSE 443
+#EXPOSE 80
+#EXPOSE 443
 
 FROM mcr.microsoft.com/dotnet/sdk:6.0 AS build
 WORKDIR /src

src/Storage.App/Extensions/DependencyInjection/XNodeDependencyInjectionExtensions.cs → src/Storage.App/Extensions/DependencyInjection/RepositoryInjectionExtensions.cs

@@ -1,14 +1,21 @@
 using Buildersoft.Andy.X.Storage.Core.Abstraction.Repository.Connection;
+using Buildersoft.Andy.X.Storage.Core.Abstraction.Repository.Consumers;
 using Buildersoft.Andy.X.Storage.Core.Repository.Connection;
+using Buildersoft.Andy.X.Storage.Core.Repository.Consumers;
 using Microsoft.Extensions.DependencyInjection;
 
 namespace Buildersoft.Andy.X.Storage.App.Extensions.DependencyInjection
 {
- public static class XNodeDependencyInjectionExtensions
+ public static class RepositoryInjectionExtensions
  {
  public static void AddNodeServiceRepository(this IServiceCollection services)
  {
  services.AddSingleton<IXNodeConnectionRepository, XNodeConnectionRepository>();
  }
+
+ public static void AddConsumerConnectionRepository(this IServiceCollection services)
+ {
+ services.AddSingleton<IConsumerConnectionRepository, ConsumerConnectionRepository>();
+ }
  }
 }

src/Storage.App/Program.cs

@@ -25,9 +25,6 @@ public static void Main(string[] args)
  if (Environment.GetEnvironmentVariable("ANDYX_CERTIFICATE_DEFAULT_PATH") != null)
  Environment.SetEnvironmentVariable("ASPNETCORE_Kestrel__Certificates__Default__Path", Environment.GetEnvironmentVariable("ANDYX_CERTIFICATE_DEFAULT_PATH"));
 
- if (Environment.GetEnvironmentVariable("ASPNETCORE_URLS") == "")
- Environment.SetEnvironmentVariable("ASPNETCORE_URLS", "https://+:443;http://+:80");
-
 
  try
  {

src/Storage.App/Startup.cs

@@ -27,6 +27,7 @@ public void ConfigureServices(IServiceCollection services)
  services.AddConfigurations(Configuration);
  services.AddSerilogLoggingConfiguration(Configuration);
  services.AddNodeServiceRepository();
+ services.AddConsumerConnectionRepository();
  services.AddIOServices();
  services.AddStartService();
 

src/Storage.App/Storage.App.csproj

@@ -5,13 +5,14 @@
  <UserSecretsId>7e4fa6b7-add2-44ec-a908-b95747757b49</UserSecretsId>
  <DockerDefaultTargetOS>Linux</DockerDefaultTargetOS>
  <RootNamespace>Buildersoft.Andy.X.Storage.App</RootNamespace>
- <Version>2.0.0</Version>
+ <Version>2.1.0</Version>
  <Company>Buildersoft</Company>
  <Product>Buildersoft Andy</Product>
  <Authors>Buildersoft</Authors>
  <Description>Buildersoft Andy X is a distributed messaging system. This system will empower developers to move into Event Driven Systems. Andy X is a multi-tenant system.</Description>
  <Copyright>Copyright © Buildersoft 2022</Copyright>
  <DockerfileContext>..\..</DockerfileContext>
+ <SignAssembly>True</SignAssembly>
 
  </PropertyGroup>
 

src/Storage.App/appsettings.json

@@ -20,11 +20,12 @@
 
  "XNodes": [
  {
- "ServiceUrl": "https://localhost:6541",
+ "ServiceUrl": "http://localhost:6540",
  "Subscription": 1,
  "JwtToken": "na",
  "Username": "admin",
- "Password": "admin"
+ "Password": "admin",
+ "SkipCertificate": false
  //"CertificateFile": "CERT_FILE",
  //"CertificatePassword": "CERT_PASSWORD"
  }
@@ -43,6 +44,7 @@
 
  "Partition": {
  "SizeInMemory": 3000,
+ "BatchSize": 3000,
  "FlushInterval": 5000,
  "PointerAcknowledgedMessageArchivationInterval": 3600000
  //"PointerAcknowledgedMessageArchivationInterval": 180000

src/Storage.Core/Abstraction/Repository/Consumers/IConsumerConnectionRepository.cs

@@ -0,0 +1,13 @@
+using Buildersoft.Andy.X.Storage.Model.App.Consumers;
+
+namespace Buildersoft.Andy.X.Storage.Core.Abstraction.Repository.Consumers
+{
+ public interface IConsumerConnectionRepository
+ {
+ Consumer GetConsumerById(string id);
+ void AddConsumer(string id, Consumer consumer);
+ void AddConsumerConnection(string id);
+ void RemoveConsumer(string id);
+ void RemoveConsumerConnection(string id);
+ }
+}

src/Storage.Core/Provider/XNodeConnectionProvider.cs

@@ -29,6 +29,7 @@ public XNodeConnectionProvider(XNodeConfiguration nodeConfig,
  this.dataStorageConfig = dataStorageConfig;
  this.agentConfiguration = agentConfiguration;
  this.agentId = agentId;
+
  ConnectToXNode();
  }
 
@@ -66,12 +67,19 @@ private void BuildConnectionWithAgent(XNodeConfiguration nodeConfig)
  {
  if (message is HttpClientHandler httpClientHandler)
  {
- httpClientHandler.ClientCertificateOptions = ClientCertificateOption.Manual;
- httpClientHandler.SslProtocols = SslProtocols.Tls12;
- var certLocation = Path.Combine(SystemLocations.GetConfigCertificateDirectory(), nodeConfig.CertificateFile);
- httpClientHandler.ClientCertificates.Add(new X509Certificate2(certLocation, nodeConfig.CertificatePassword));
+ if (nodeConfig.SkipCertificate == true)
+ {
+ httpClientHandler.ServerCertificateCustomValidationCallback +=
+ (sender, certificate, chain, sslPolicyErrors) => { return true; };
+ }
+ else
+ {
+ httpClientHandler.ClientCertificateOptions = ClientCertificateOption.Manual;
+ httpClientHandler.SslProtocols = SslProtocols.Tls12;
+ var certLocation = Path.Combine(SystemLocations.GetConfigCertificateDirectory(), nodeConfig.CertificateFile);
+ httpClientHandler.ClientCertificates.Add(new X509Certificate2(certLocation, nodeConfig.CertificatePassword));
+ }
  }
-
  return message;
  };
  }

src/Storage.Core/Repository/Consumers/ConsumerConnectionRepository.cs

@@ -0,0 +1,53 @@
+using System;
+using System.Collections.Concurrent;
+using Buildersoft.Andy.X.Storage.Core.Abstraction.Repository.Consumers;
+using Buildersoft.Andy.X.Storage.Model.App.Consumers;
+
+namespace Buildersoft.Andy.X.Storage.Core.Repository.Consumers
+{
+ public class ConsumerConnectionRepository : IConsumerConnectionRepository
+ {
+ private readonly ConcurrentDictionary<string, Consumer> _consumersConnected;
+
+ public ConsumerConnectionRepository()
+ {
+ _consumersConnected = new ConcurrentDictionary<string, Consumer>();
+ }
+
+ public Consumer GetConsumerById(string id)
+ {
+ return _consumersConnected.ContainsKey(id)
+ ? _consumersConnected[id]
+ : null;
+ }
+
+ public void AddConsumer(string id, Consumer consumer)
+ {
+ _consumersConnected.TryAdd(id, consumer);
+ }
+
+ public void AddConsumerConnection(string id)
+ {
+ if (_consumersConnected.ContainsKey(id))
+ _consumersConnected[id].Connections.Add(Guid.NewGuid());
+ }
+
+ public void RemoveConsumer(string id)
+ {
+ if (!_consumersConnected.ContainsKey(id))
+ return;
+
+ if (_consumersConnected[id].Connections.Count == 0)
+ _consumersConnected.TryRemove(id, out _);
+ }
+
+ public void RemoveConsumerConnection(string id)
+ {
+ if (!_consumersConnected.ContainsKey(id))
+ return;
+
+ if (_consumersConnected[id].Connections.Count != 0)
+ _consumersConnected[id].Connections.RemoveAt(0);
+ }
+ }
+}

src/Storage.Core/Service/System/SystemService.cs

@@ -1,4 +1,5 @@
 using Buildersoft.Andy.X.Storage.Core.Abstraction.Repository.Connection;
+using Buildersoft.Andy.X.Storage.Core.Abstraction.Repository.Consumers;
 using Buildersoft.Andy.X.Storage.Core.Service.XNodes;
 using Buildersoft.Andy.X.Storage.IO.Locations;
 using Buildersoft.Andy.X.Storage.IO.Readers;
@@ -22,7 +23,9 @@ public class SystemService
  private readonly TenantIOService _tenantIOService;
  private readonly ProducerIOService _producerIOService;
  private readonly ConsumerIOService _consumerIOService;
- private readonly MessageIOService _messageIOService2;
+ private readonly MessageIOService _messageIOService;
+ private readonly IConsumerConnectionRepository _consumerConnectionRepository;
+
  private readonly List<XNodeConfiguration> nodes;
  private readonly DataStorageConfiguration dataStorage;
  private readonly AgentConfiguration agent;
@@ -38,7 +41,8 @@ public SystemService(
  TenantIOService tenantIOService,
  ProducerIOService producerIOService,
  ConsumerIOService consumerIOService,
- MessageIOService messageIOService2)
+ MessageIOService messageIOService,
+ IConsumerConnectionRepository consumerConnectionRepository)
  {
  _logger = logger;
  _serviceProvider = serviceProvider;
@@ -48,7 +52,9 @@ public SystemService(
  _tenantIOService = tenantIOService;
  _producerIOService = producerIOService;
  _consumerIOService = consumerIOService;
- _messageIOService2 = messageIOService2;
+ _messageIOService = messageIOService;
+ _consumerConnectionRepository = consumerConnectionRepository;
+
  nodes = _serviceProvider.GetService(typeof(List<XNodeConfiguration>)) as List<XNodeConfiguration>;
  dataStorage = _serviceProvider.GetService(typeof(DataStorageConfiguration)) as DataStorageConfiguration;
  agent = _serviceProvider.GetService(typeof(AgentConfiguration)) as AgentConfiguration;
@@ -136,12 +142,14 @@ private void InitializeServices()
  agentId,
  xnode,
  dataStorage,
+ partition,
  agent,
  _xNodeConnectionRepository,
  _tenantIOService,
  _producerIOService,
  _consumerIOService,
- _messageIOService2);
+ _messageIOService,
+ _consumerConnectionRepository);
  }
  }
  else