Separate websocket manager to rabbitmq consumer, fixed bug in creating user function
This commit is contained in:
parent
9840e0c9f8
commit
30e4991032
|
@ -14,7 +14,7 @@ I'll reroute my emails to one of you for software updates.
|
|||
|
||||
|
||||
MARIOS: Please make sure to patch out the vulnerable npm packages in the frontend.
|
||||
And in my opinion, get started on React or Testcafe Integration tests ;)
|
||||
Get started on React or Testcafe Integration tests ;)
|
||||
You can add them into the Gitea Actions Pipeline, read the documentation on Github-actions and integration tests.
|
||||
|
||||
Runner:
|
||||
|
|
|
@ -326,11 +326,20 @@ public class Controller : ControllerBase
|
|||
[HttpPost(nameof(CreateUser))]
|
||||
public async Task<ActionResult<User>> CreateUser([FromBody] User newUser, Token authToken)
|
||||
{
|
||||
var create = Db.GetSession(authToken).Create(newUser);
|
||||
|
||||
return create && await Db.SendNewUserEmail(newUser)
|
||||
? newUser.HidePassword()
|
||||
: Unauthorized() ;
|
||||
var create = Db.GetSession(authToken).Create(newUser);
|
||||
if (create)
|
||||
{
|
||||
var mail_success= await Db.SendNewUserEmail(newUser);
|
||||
if (!mail_success)
|
||||
{
|
||||
Db.GetSession(authToken).Delete(newUser);
|
||||
}
|
||||
|
||||
return mail_success ? newUser.HidePassword():Unauthorized();
|
||||
}
|
||||
|
||||
return Unauthorized() ;
|
||||
}
|
||||
|
||||
[HttpPost(nameof(CreateInstallation))]
|
||||
|
|
|
@ -22,18 +22,10 @@ public static class Program
|
|||
Db.Init();
|
||||
var builder = WebApplication.CreateBuilder(args);
|
||||
|
||||
|
||||
string vpnServerIp = "194.182.190.208";
|
||||
//string vpnServerIp = "127.0.0.1";
|
||||
WebsocketManager.Factory = new ConnectionFactory { HostName = vpnServerIp};
|
||||
WebsocketManager.Connection = WebsocketManager.Factory.CreateConnection();
|
||||
WebsocketManager.Channel = WebsocketManager.Connection.CreateModel();
|
||||
Console.WriteLine("Middleware subscribed to RabbitMQ queue, ready for receiving messages");
|
||||
WebsocketManager.Channel.QueueDeclare(queue: "statusQueue", durable: true, exclusive: false, autoDelete: false, arguments: null);
|
||||
|
||||
WebsocketManager.StartRabbitMqConsumer();
|
||||
RabbitMqManager.InitializeEnvironment();
|
||||
RabbitMqManager.StartRabbitMqConsumer();
|
||||
Console.WriteLine("Queue declared");
|
||||
WebsocketManager.InformInstallationsToSubscribeToRabbitMq();
|
||||
//WebsocketManager.InformInstallationsToSubscribeToRabbitMq();
|
||||
WebsocketManager.MonitorInstallationTable();
|
||||
|
||||
builder.Services.AddControllers();
|
||||
|
|
|
@ -0,0 +1,174 @@
|
|||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using InnovEnergy.App.Backend.Database;
|
||||
using InnovEnergy.App.Backend.DataTypes;
|
||||
using RabbitMQ.Client;
|
||||
using RabbitMQ.Client.Events;
|
||||
|
||||
namespace InnovEnergy.App.Backend.Websockets;
|
||||
|
||||
public static class RabbitMqManager
|
||||
{
|
||||
|
||||
public static ConnectionFactory Factory = null!;
|
||||
public static IConnection Connection = null!;
|
||||
public static IModel Channel = null!;
|
||||
|
||||
|
||||
public static void InitializeEnvironment()
|
||||
{
|
||||
|
||||
string vpnServerIp = "194.182.190.208";
|
||||
Factory = new ConnectionFactory { HostName = vpnServerIp};
|
||||
Connection = Factory.CreateConnection();
|
||||
Channel = Connection.CreateModel();
|
||||
Console.WriteLine("Middleware subscribed to RabbitMQ queue, ready for receiving messages");
|
||||
Channel.QueueDeclare(queue: "statusQueue", durable: true, exclusive: false, autoDelete: false, arguments: null);
|
||||
}
|
||||
|
||||
public static async Task StartRabbitMqConsumer()
|
||||
{
|
||||
var consumer = new EventingBasicConsumer(Channel);
|
||||
consumer.Received += (_, ea) =>
|
||||
{
|
||||
var body = ea.Body.ToArray();
|
||||
var message = Encoding.UTF8.GetString(body);
|
||||
StatusMessage? receivedStatusMessage = JsonSerializer.Deserialize<StatusMessage>(message);
|
||||
var InstallationConnections = WebsocketManager.InstallationConnections;
|
||||
|
||||
lock (InstallationConnections)
|
||||
{
|
||||
//Consumer received a message
|
||||
if (receivedStatusMessage != null)
|
||||
{
|
||||
Console.WriteLine("----------------------------------------------");
|
||||
Console.WriteLine("Received a message from installation: " + receivedStatusMessage.InstallationId + " and status is: " + receivedStatusMessage.Status);
|
||||
var installationId = receivedStatusMessage.InstallationId;
|
||||
|
||||
//This is a heartbit message, just update the timestamp for this installation.
|
||||
//There is no need to notify the corresponding front-ends.
|
||||
if (receivedStatusMessage.Type == MessageType.Heartbit)
|
||||
{
|
||||
Console.WriteLine("This is a heartbit message from installation: " + installationId);
|
||||
}
|
||||
else
|
||||
{
|
||||
//Traverse the Warnings list, and store each of them to the database
|
||||
if (receivedStatusMessage.Warnings != null)
|
||||
{
|
||||
foreach (var warning in receivedStatusMessage.Warnings)
|
||||
{
|
||||
Warning newWarning = new Warning
|
||||
{
|
||||
InstallationId = receivedStatusMessage.InstallationId,
|
||||
Description = warning.Description,
|
||||
Date = warning.Date,
|
||||
Time = warning.Time,
|
||||
DeviceCreatedTheMessage = warning.CreatedBy,
|
||||
Seen = false
|
||||
};
|
||||
//Create a new warning and add it to the database
|
||||
Db.HandleWarning(newWarning, receivedStatusMessage.InstallationId);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
//Traverse the Alarm list, and store each of them to the database
|
||||
if (receivedStatusMessage.Alarms != null)
|
||||
{
|
||||
foreach (var alarm in receivedStatusMessage.Alarms)
|
||||
{
|
||||
Error newError = new Error
|
||||
{
|
||||
InstallationId = receivedStatusMessage.InstallationId,
|
||||
Description = alarm.Description,
|
||||
Date = alarm.Date,
|
||||
Time = alarm.Time,
|
||||
DeviceCreatedTheMessage = alarm.CreatedBy,
|
||||
Seen = false
|
||||
};
|
||||
//Create a new error and add it to the database
|
||||
Db.HandleError(newError, receivedStatusMessage.InstallationId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
//This installation id does not exist in our data structure, add it.
|
||||
if (!InstallationConnections.ContainsKey(installationId))
|
||||
{
|
||||
Console.WriteLine("Create new empty list for installation: " + installationId);
|
||||
InstallationConnections[installationId] = new InstallationInfo
|
||||
{
|
||||
Status = receivedStatusMessage.Status,
|
||||
Timestamp = DateTime.Now
|
||||
};
|
||||
}
|
||||
else
|
||||
{
|
||||
InstallationConnections[installationId].Status = receivedStatusMessage.Status;
|
||||
InstallationConnections[installationId].Timestamp = DateTime.Now;
|
||||
}
|
||||
|
||||
//Console.WriteLine("----------------------------------------------");
|
||||
//Update all the connected front-ends regarding this installation
|
||||
if(InstallationConnections[installationId].Connections.Count > 0)
|
||||
{
|
||||
WebsocketManager.InformWebsocketsForInstallation(installationId);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
};
|
||||
Channel.BasicConsume(queue: "statusQueue", autoAck: true, consumer: consumer);
|
||||
}
|
||||
|
||||
public static void InformInstallationsToSubscribeToRabbitMq()
|
||||
{
|
||||
var installationIps = Db.Installations.Select(inst => inst.VpnIp).ToList();
|
||||
Console.WriteLine("Count is "+installationIps.Count);
|
||||
var maxRetransmissions = 2;
|
||||
|
||||
UdpClient udpClient = new UdpClient();
|
||||
udpClient.Client.ReceiveTimeout = 2000;
|
||||
int port = 9000;
|
||||
//Send a message to each installation and tell it to subscribe to the queue
|
||||
using (udpClient)
|
||||
{
|
||||
for (int i = 0; i < installationIps.Count; i++)
|
||||
{
|
||||
if(installationIps[i]==""){continue;}
|
||||
Console.WriteLine("-----------------------------------------------------------");
|
||||
Console.WriteLine("Trying to reach installation with IP: " + installationIps[i]);
|
||||
//Try at most MAX_RETRANSMISSIONS times to reach an installation.
|
||||
for (int j = 0; j < maxRetransmissions; j++)
|
||||
{
|
||||
string message = "This is a message from RabbitMQ server, you can subscribe to the RabbitMQ queue";
|
||||
byte[] data = Encoding.UTF8.GetBytes(message);
|
||||
udpClient.Send(data, data.Length, installationIps[i], port);
|
||||
|
||||
Console.WriteLine($"Sent UDP message to {installationIps[i]}:{port}: {message}");
|
||||
IPEndPoint remoteEndPoint = new IPEndPoint(IPAddress.Parse(installationIps[i]), port);
|
||||
|
||||
try
|
||||
{
|
||||
byte[] replyData = udpClient.Receive(ref remoteEndPoint);
|
||||
string replyMessage = Encoding.UTF8.GetString(replyData);
|
||||
Console.WriteLine("Received " + replyMessage + " from installation " + installationIps[i]);
|
||||
break;
|
||||
}
|
||||
catch (SocketException ex)
|
||||
{
|
||||
if (ex.SocketErrorCode == SocketError.TimedOut){Console.WriteLine("Timed out waiting for a response. Retry...");}
|
||||
else{Console.WriteLine("Error: " + ex.Message);}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Console.WriteLine("Start RabbitMQ Consumer");
|
||||
|
||||
}
|
||||
|
||||
}
|
|
@ -1,68 +1,12 @@
|
|||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Net.WebSockets;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using InnovEnergy.App.Backend.Database;
|
||||
using InnovEnergy.App.Backend.DataTypes;
|
||||
using RabbitMQ.Client;
|
||||
using RabbitMQ.Client.Events;
|
||||
|
||||
namespace InnovEnergy.App.Backend.Websockets;
|
||||
|
||||
public static class WebsocketManager
|
||||
{
|
||||
public static Dictionary<int, InstallationInfo> InstallationConnections = new Dictionary<int, InstallationInfo>();
|
||||
public static ConnectionFactory Factory = null!;
|
||||
public static IConnection Connection = null!;
|
||||
public static IModel Channel = null!;
|
||||
|
||||
public static void InformInstallationsToSubscribeToRabbitMq()
|
||||
{
|
||||
var installationIps = Db.Installations.Select(inst => inst.VpnIp).ToList();
|
||||
Console.WriteLine("Count is "+installationIps.Count);
|
||||
var maxRetransmissions = 2;
|
||||
|
||||
UdpClient udpClient = new UdpClient();
|
||||
udpClient.Client.ReceiveTimeout = 2000;
|
||||
int port = 9000;
|
||||
//Send a message to each installation and tell it to subscribe to the queue
|
||||
using (udpClient)
|
||||
{
|
||||
for (int i = 0; i < installationIps.Count; i++)
|
||||
{
|
||||
if(installationIps[i]==""){continue;}
|
||||
Console.WriteLine("-----------------------------------------------------------");
|
||||
Console.WriteLine("Trying to reach installation with IP: " + installationIps[i]);
|
||||
//Try at most MAX_RETRANSMISSIONS times to reach an installation.
|
||||
for (int j = 0; j < maxRetransmissions; j++)
|
||||
{
|
||||
string message = "This is a message from RabbitMQ server, you can subscribe to the RabbitMQ queue";
|
||||
byte[] data = Encoding.UTF8.GetBytes(message);
|
||||
udpClient.Send(data, data.Length, installationIps[i], port);
|
||||
|
||||
Console.WriteLine($"Sent UDP message to {installationIps[i]}:{port}: {message}");
|
||||
IPEndPoint remoteEndPoint = new IPEndPoint(IPAddress.Parse(installationIps[i]), port);
|
||||
|
||||
try
|
||||
{
|
||||
byte[] replyData = udpClient.Receive(ref remoteEndPoint);
|
||||
string replyMessage = Encoding.UTF8.GetString(replyData);
|
||||
Console.WriteLine("Received " + replyMessage + " from installation " + installationIps[i]);
|
||||
break;
|
||||
}
|
||||
catch (SocketException ex)
|
||||
{
|
||||
if (ex.SocketErrorCode == SocketError.TimedOut){Console.WriteLine("Timed out waiting for a response. Retry...");}
|
||||
else{Console.WriteLine("Error: " + ex.Message);}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Console.WriteLine("Start RabbitMQ Consumer");
|
||||
|
||||
}
|
||||
|
||||
//Every 1 minute, check the timestamp of the latest received message for every installation.
|
||||
//If the difference between the two timestamps is more than one minute, we consider this installation unavailable.
|
||||
|
@ -82,100 +26,6 @@ public static class WebsocketManager
|
|||
}
|
||||
}
|
||||
|
||||
public static async Task StartRabbitMqConsumer()
|
||||
{
|
||||
var consumer = new EventingBasicConsumer(Channel);
|
||||
consumer.Received += (_, ea) =>
|
||||
{
|
||||
var body = ea.Body.ToArray();
|
||||
var message = Encoding.UTF8.GetString(body);
|
||||
StatusMessage? receivedStatusMessage = JsonSerializer.Deserialize<StatusMessage>(message);
|
||||
|
||||
lock (InstallationConnections)
|
||||
{
|
||||
//Consumer received a message
|
||||
if (receivedStatusMessage != null)
|
||||
{
|
||||
Console.WriteLine("Received a message from installation: " + receivedStatusMessage.InstallationId + " and status is: " + receivedStatusMessage.Status);
|
||||
Console.WriteLine("----------------------------------------------");
|
||||
Console.WriteLine("Update installation connection table");
|
||||
var installationId = receivedStatusMessage.InstallationId;
|
||||
|
||||
//This is a heartbit message, just update the timestamp for this installation.
|
||||
//There is no need to notify the corresponding front-ends.
|
||||
if (receivedStatusMessage.Type == MessageType.Heartbit)
|
||||
{
|
||||
InstallationConnections[installationId].Timestamp = DateTime.Now;
|
||||
}
|
||||
else
|
||||
{
|
||||
//Traverse the Warnings list, and store each of them to the database
|
||||
if (receivedStatusMessage.Warnings != null)
|
||||
{
|
||||
foreach (var warning in receivedStatusMessage.Warnings)
|
||||
{
|
||||
Warning newWarning = new Warning
|
||||
{
|
||||
InstallationId = receivedStatusMessage.InstallationId,
|
||||
Description = warning.Description,
|
||||
Date = warning.Date,
|
||||
Time = warning.Time,
|
||||
DeviceCreatedTheMessage = warning.CreatedBy,
|
||||
Seen = false
|
||||
};
|
||||
//Create a new warning and add it to the database
|
||||
Db.HandleWarning(newWarning, receivedStatusMessage.InstallationId);
|
||||
}
|
||||
}
|
||||
|
||||
//Traverse the Alarm list, and store each of them to the database
|
||||
if (receivedStatusMessage.Alarms != null)
|
||||
{
|
||||
foreach (var alarm in receivedStatusMessage.Alarms)
|
||||
{
|
||||
Error newError = new Error
|
||||
{
|
||||
InstallationId = receivedStatusMessage.InstallationId,
|
||||
Description = alarm.Description,
|
||||
Date = alarm.Date,
|
||||
Time = alarm.Time,
|
||||
DeviceCreatedTheMessage = alarm.CreatedBy,
|
||||
Seen = false
|
||||
};
|
||||
//Create a new error and add it to the database
|
||||
Db.HandleError(newError, receivedStatusMessage.InstallationId);
|
||||
}
|
||||
}
|
||||
|
||||
//This installation id does not exist in our data structure, add it.
|
||||
if (!InstallationConnections.ContainsKey(installationId))
|
||||
{
|
||||
Console.WriteLine("Create new empty list for installation: " + installationId);
|
||||
InstallationConnections[installationId] = new InstallationInfo
|
||||
{
|
||||
Status = receivedStatusMessage.Status,
|
||||
Timestamp = DateTime.Now
|
||||
};
|
||||
}
|
||||
else
|
||||
{
|
||||
InstallationConnections[installationId].Status = receivedStatusMessage.Status;
|
||||
InstallationConnections[installationId].Timestamp = DateTime.Now;
|
||||
}
|
||||
|
||||
Console.WriteLine("----------------------------------------------");
|
||||
//Update all the connected front-ends regarding this installation
|
||||
if(InstallationConnections[installationId].Connections.Count > 0)
|
||||
{
|
||||
InformWebsocketsForInstallation(installationId);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
Channel.BasicConsume(queue: "statusQueue", autoAck: true, consumer: consumer);
|
||||
}
|
||||
|
||||
//Inform all the connected websockets regarding installation "installationId"
|
||||
public static void InformWebsocketsForInstallation(int installationId)
|
||||
{
|
||||
|
|
|
@ -1,9 +0,0 @@
|
|||
using System.Net.WebSockets;
|
||||
|
||||
namespace InnovEnergy.App.Middleware;
|
||||
|
||||
public class InstallationInfo
|
||||
{
|
||||
public int Status { get; set; }
|
||||
public List<WebSocket> Connections { get; } = new List<WebSocket>();
|
||||
}
|
|
@ -1,18 +0,0 @@
|
|||
<Project Sdk="Microsoft.NET.Sdk.Web">
|
||||
|
||||
<Import Project="../InnovEnergy.App.props" />
|
||||
|
||||
<PropertyGroup>
|
||||
<RootNamespace>InnovEnergy.App.Middleware</RootNamespace>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="RabbitMQ.Client" Version="6.6.0" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\Lib\Utils\Utils.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
|
||||
</Project>
|
|
@ -1,66 +0,0 @@
|
|||
using InnovEnergy.App.Middleware;
|
||||
using System;
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Net.WebSockets;
|
||||
using System.Text;
|
||||
using InnovEnergy.Lib.Utils;
|
||||
|
||||
internal class Program
|
||||
{
|
||||
public static readonly object SharedDataLock = new object();
|
||||
|
||||
public static async Task Main(string[] args)
|
||||
{
|
||||
//For each installation id, we maintain a list of the connected clients
|
||||
var installationConnections = new Dictionary<int, InstallationInfo>();
|
||||
var installationsIds = new List<int> {1};
|
||||
var installationIps = new List<string> {"10.2.3.115"};
|
||||
var MAX_RETRANSMISSIONS = 2;
|
||||
|
||||
RabbitMqConsumer.StartRabbitMqConsumer(installationConnections,SharedDataLock);
|
||||
|
||||
UdpClient udpClient = new UdpClient();
|
||||
udpClient.Client.ReceiveTimeout = 2000;
|
||||
int port = 9000;
|
||||
//Send a message to each installation and tell it to subscribe to the queue
|
||||
for (int i = 0; i < installationsIds.Count; i++)
|
||||
{
|
||||
using (udpClient)
|
||||
{
|
||||
//Try at most MAX_RETRANSMISSIONS times to reach an installation.
|
||||
for (int j = 0; j < MAX_RETRANSMISSIONS; j++)
|
||||
{
|
||||
string message = "This is a message from RabbitMQ server, you can subscribe to the RabbitMQ queue";
|
||||
byte[] data = Encoding.UTF8.GetBytes(message);
|
||||
udpClient.Send(data, data.Length, installationIps[i], port);
|
||||
|
||||
Console.WriteLine($"Sent UDP message to {installationIps[i]}:{port}: {message}");
|
||||
IPEndPoint remoteEndPoint = new IPEndPoint(IPAddress.Parse(installationIps[i]), port);
|
||||
|
||||
try
|
||||
{
|
||||
byte[] replyData = udpClient.Receive(ref remoteEndPoint);
|
||||
string replyMessage = Encoding.UTF8.GetString(replyData);
|
||||
Console.WriteLine("Received message from installation " + installationsIds[i]);
|
||||
break;
|
||||
}
|
||||
catch (SocketException ex)
|
||||
{
|
||||
if (ex.SocketErrorCode == SocketError.TimedOut)
|
||||
{
|
||||
Console.WriteLine("Timed out waiting for a response. Retry...");
|
||||
}
|
||||
else
|
||||
{
|
||||
Console.WriteLine("Error: " + ex.Message);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Console.WriteLine("WebSocket server is running. Press Enter to exit.");
|
||||
await WebSocketListener.StartServerAsync(installationConnections,SharedDataLock);
|
||||
}
|
||||
}
|
|
@ -1,88 +0,0 @@
|
|||
using System.Net.WebSockets;
|
||||
using System.Text.Json;
|
||||
namespace InnovEnergy.App.Middleware;
|
||||
using System.Text;
|
||||
using RabbitMQ.Client;
|
||||
using RabbitMQ.Client.Events;
|
||||
|
||||
public static class RabbitMqConsumer
|
||||
{
|
||||
|
||||
private static ConnectionFactory _factory = null!;
|
||||
private static IConnection _connection = null!;
|
||||
private static IModel _channel= null!;
|
||||
|
||||
public static void StartRabbitMqConsumer(Dictionary<Int32, InstallationInfo> installationConnections, Object sharedDataLock)
|
||||
{
|
||||
string vpnServerIp = "194.182.190.208";
|
||||
_factory = new ConnectionFactory { HostName = "localhost" };
|
||||
_connection = _factory.CreateConnection();
|
||||
_channel = _connection.CreateModel();
|
||||
Console.WriteLine("Middleware subscribed to RabbitMQ queue, ready for receiving messages");
|
||||
_channel.QueueDeclare(queue: "statusQueue", durable: false, exclusive: false, autoDelete: false, arguments: null);
|
||||
|
||||
var consumer = new EventingBasicConsumer(_channel);
|
||||
consumer.Received += (_, ea) => Callback(installationConnections, sharedDataLock, ea);
|
||||
|
||||
_channel.BasicConsume(queue: "statusQueue", autoAck: true, consumer: consumer);
|
||||
|
||||
}
|
||||
|
||||
private static void Callback(Dictionary<Int32, InstallationInfo> installationConnections, Object sharedDataLock, BasicDeliverEventArgs ea)
|
||||
{
|
||||
var body = ea.Body.ToArray();
|
||||
var message = Encoding.UTF8.GetString(body);
|
||||
StatusMessage? receivedStatusMessage = JsonSerializer.Deserialize<StatusMessage>(message);
|
||||
|
||||
lock (sharedDataLock)
|
||||
{
|
||||
// Process the received message
|
||||
if (receivedStatusMessage != null)
|
||||
{
|
||||
Console.WriteLine("Received a message from installation: " + receivedStatusMessage.InstallationId + " and status is: " + receivedStatusMessage.Status);
|
||||
Console.WriteLine("----------------------------------------------");
|
||||
Console.WriteLine("Update installation connection table");
|
||||
var installationId = receivedStatusMessage.InstallationId;
|
||||
|
||||
if (!installationConnections.ContainsKey(installationId))
|
||||
{
|
||||
Console.WriteLine("Create new empty list for installation: " + installationId);
|
||||
installationConnections[installationId] = new InstallationInfo
|
||||
{
|
||||
Status = receivedStatusMessage.Status
|
||||
};
|
||||
}
|
||||
|
||||
Console.WriteLine("----------------------------------------------");
|
||||
|
||||
foreach (var installationConnection in installationConnections)
|
||||
{
|
||||
if (installationConnection.Key == installationId && installationConnection.Value.Connections.Count > 0)
|
||||
{
|
||||
Console.WriteLine("Update all the connected websockets for installation " + installationId);
|
||||
installationConnection.Value.Status = receivedStatusMessage.Status;
|
||||
|
||||
var jsonObject = new
|
||||
{
|
||||
id = installationId,
|
||||
status = receivedStatusMessage.Status
|
||||
};
|
||||
|
||||
string jsonString = JsonSerializer.Serialize(jsonObject);
|
||||
byte[] dataToSend = Encoding.UTF8.GetBytes(jsonString);
|
||||
|
||||
foreach (var connection in installationConnection.Value.Connections)
|
||||
{
|
||||
connection.SendAsync(
|
||||
new ArraySegment<byte>(dataToSend, 0, dataToSend.Length),
|
||||
WebSocketMessageType.Text,
|
||||
true, // Indicates that this is the end of the message
|
||||
CancellationToken.None
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
|
@ -1,8 +0,0 @@
|
|||
|
||||
namespace InnovEnergy.App.Middleware;
|
||||
|
||||
public class StatusMessage
|
||||
{
|
||||
public required int InstallationId { get; init; }
|
||||
public required int Status { get; init; }
|
||||
}
|
|
@ -1,129 +0,0 @@
|
|||
using System.Net;
|
||||
using System.Net.WebSockets;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
|
||||
namespace InnovEnergy.App.Middleware;
|
||||
|
||||
public static class WebSocketListener
|
||||
{
|
||||
|
||||
public static async Task StartServerAsync(Dictionary<Int32, InstallationInfo> installationConnections, Object sharedDataLock)
|
||||
{
|
||||
var listener = new HttpListener();
|
||||
listener.Prefixes.Add("http://127.0.0.1:8080/");
|
||||
|
||||
listener.Start();
|
||||
|
||||
//Http listener listens for connections. When it accepts a new connection, it creates a new Task to handle this connection
|
||||
while (true)
|
||||
{
|
||||
var context = await listener.GetContextAsync();
|
||||
if (context.Request.IsWebSocketRequest)
|
||||
{
|
||||
var webSocketContext = await context.AcceptWebSocketAsync(null);
|
||||
var webSocket = webSocketContext.WebSocket;
|
||||
|
||||
// Add the connected WebSocket to the collection
|
||||
Console.WriteLine("Accepted a new websocket connection");
|
||||
HandleWebSocketConnection(webSocket, installationConnections);
|
||||
}
|
||||
else
|
||||
{
|
||||
context.Response.StatusCode = 400;
|
||||
context.Response.Close();
|
||||
}
|
||||
}
|
||||
|
||||
//We have a task per websocket connection
|
||||
async Task HandleWebSocketConnection(WebSocket currentWebSocket, Dictionary<Int32, InstallationInfo> installationConnections)
|
||||
{
|
||||
|
||||
var buffer = new byte[4096];
|
||||
|
||||
try
|
||||
{
|
||||
while (currentWebSocket.State == WebSocketState.Open)
|
||||
{
|
||||
//Listen for incoming messages on this WebSocket
|
||||
var result = await currentWebSocket.ReceiveAsync(buffer, CancellationToken.None);
|
||||
Console.WriteLine("Received a new message from websocket");
|
||||
if (result.MessageType != WebSocketMessageType.Text)
|
||||
continue;
|
||||
|
||||
var message = Encoding.UTF8.GetString(buffer, 0, result.Count);
|
||||
var installationIds = JsonSerializer.Deserialize<int[]>(message);
|
||||
|
||||
lock (sharedDataLock)
|
||||
{
|
||||
//Each front-end will send the list of the installations it wants to access
|
||||
//If this is a new key (installation id), initialize the list for this key and then add the websocket object for this client
|
||||
//Then, report the status of each requested installation to the front-end that created the websocket connection
|
||||
foreach (var installationId in installationIds)
|
||||
{
|
||||
if (!installationConnections.ContainsKey(installationId))
|
||||
{
|
||||
Console.WriteLine("Create new empty list for installation id "+installationId);
|
||||
installationConnections[installationId] = new InstallationInfo
|
||||
{
|
||||
Status = -2
|
||||
};
|
||||
}
|
||||
|
||||
installationConnections[installationId].Connections.Add(currentWebSocket);
|
||||
|
||||
var jsonObject = new
|
||||
{
|
||||
id = installationId,
|
||||
status = installationConnections[installationId].Status
|
||||
};
|
||||
|
||||
var jsonString = JsonSerializer.Serialize(jsonObject);
|
||||
var dataToSend = Encoding.UTF8.GetBytes(jsonString);
|
||||
|
||||
|
||||
currentWebSocket.SendAsync(dataToSend,
|
||||
WebSocketMessageType.Text,
|
||||
true, // Indicates that this is the end of the message
|
||||
CancellationToken.None
|
||||
);
|
||||
}
|
||||
|
||||
Console.WriteLine("Printing installation connection list");
|
||||
Console.WriteLine("----------------------------------------------");
|
||||
foreach (var installationConnection in installationConnections)
|
||||
{
|
||||
Console.WriteLine("Installation ID: " + installationConnection.Key + " Number of Connections: " + installationConnection.Value.Connections.Count);
|
||||
}
|
||||
Console.WriteLine("----------------------------------------------");
|
||||
}
|
||||
}
|
||||
|
||||
//When the front-end terminates the connection, the following code will be executed
|
||||
Console.WriteLine("The connection has been terminated");
|
||||
foreach (var installationConnection in installationConnections)
|
||||
{
|
||||
if (installationConnection.Value.Connections.Contains(currentWebSocket))
|
||||
{
|
||||
installationConnection.Value.Connections.Remove(currentWebSocket);
|
||||
}
|
||||
}
|
||||
|
||||
await currentWebSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Connection closed by server", CancellationToken.None);
|
||||
//Print the installationConnections dictionary after deleting a websocket
|
||||
Console.WriteLine("Print the installation connections list after deleting a websocket");
|
||||
Console.WriteLine("----------------------------------------------");
|
||||
foreach (var installationConnection in installationConnections)
|
||||
{
|
||||
Console.WriteLine("Installation ID: "+ installationConnection.Key+" Number of Connections: "+installationConnection.Value.Connections.Count);
|
||||
}
|
||||
Console.WriteLine("----------------------------------------------");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Console.WriteLine("WebSocket error: " + ex.Message);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1 +0,0 @@
|
|||
dotnet publish Middleware.csproj -c Release -r linux-x64 --self-contained true -p:PublishTrimmed=false && rsync -av bin/Release/net6.0/linux-x64/publish/ ubuntu@194.182.190.208:~/middleware
|
|
@ -262,6 +262,7 @@ internal static class Program
|
|||
{
|
||||
var s3Bucket = Config.Load().S3?.Bucket;
|
||||
|
||||
//Every 15 iterations(30 seconds), the installation sends a heartbit message to the queue
|
||||
_heartBitInterval++;
|
||||
|
||||
//When the controller boots, it tries to subscribe to the queue
|
||||
|
@ -269,14 +270,9 @@ internal static class Program
|
|||
{
|
||||
_subscribeToQueueForTheFirstTime = true;
|
||||
SubscribeToQueue(currentSalimaxState, s3Bucket);
|
||||
|
||||
if (_subscribedToQueue && currentSalimaxState.Status != _prevSalimaxState)
|
||||
{
|
||||
_prevSalimaxState = currentSalimaxState.Status;
|
||||
}
|
||||
}
|
||||
//If already subscribed to the queue and the status has been changed, update the queue
|
||||
else if (_subscribedToQueue && currentSalimaxState.Status != _prevSalimaxState)
|
||||
if (_subscribedToQueue && currentSalimaxState.Status != _prevSalimaxState)
|
||||
{
|
||||
_prevSalimaxState = currentSalimaxState.Status;
|
||||
if (s3Bucket != null)
|
||||
|
@ -284,6 +280,7 @@ internal static class Program
|
|||
}
|
||||
else if (_subscribedToQueue && _heartBitInterval>=15)
|
||||
{
|
||||
//Send a heartbit to the backend
|
||||
Console.WriteLine("----------------------------------------Sending Heartbit----------------------------------------");
|
||||
_heartBitInterval = 0;
|
||||
currentSalimaxState.Type = MessageType.Heartbit;
|
||||
|
@ -320,11 +317,9 @@ internal static class Program
|
|||
_factory = new ConnectionFactory { HostName = VpnServerIp };
|
||||
_connection = _factory.CreateConnection();
|
||||
_channel = _connection.CreateModel();
|
||||
|
||||
_channel.QueueDeclare(queue: "statusQueue", durable: true, exclusive: false, autoDelete: false, arguments: null);
|
||||
|
||||
Console.WriteLine("The controller sends its status to the middleware for the first time");
|
||||
|
||||
if (s3Bucket != null) InformMiddleware(currentSalimaxState);
|
||||
|
||||
_subscribedToQueue = true;
|
||||
|
|
|
@ -48,8 +48,8 @@ function Configuration(props: ConfigurationProps) {
|
|||
<TextField
|
||||
label={
|
||||
<FormattedMessage
|
||||
id="calibration_charge_forced"
|
||||
defaultMessage="Calibration Charge forced"
|
||||
id="forced_calibration_charge"
|
||||
defaultMessage="Forced Calibration Charge"
|
||||
/>
|
||||
}
|
||||
value={props.values.calibrationChargeForced.values[0].value}
|
||||
|
|
|
@ -80,7 +80,7 @@ function installationForm(props: installationFormProps) {
|
|||
<Box
|
||||
sx={{
|
||||
position: 'absolute',
|
||||
top: isMobile ? '50%' : '30%',
|
||||
top: isMobile ? '50%' : '40%',
|
||||
left: '50%',
|
||||
transform: 'translate(-50%, -50%)',
|
||||
width: 500,
|
||||
|
@ -152,10 +152,11 @@ function installationForm(props: installationFormProps) {
|
|||
error={formValues.country === ''}
|
||||
/>
|
||||
</div>
|
||||
|
||||
<div>
|
||||
<TextField
|
||||
label={<FormattedMessage id="VpnIp" defaultMessage="VpnIp" />}
|
||||
name="VpnIp"
|
||||
name="vpnIp"
|
||||
value={formValues.vpnIp}
|
||||
onChange={handleChange}
|
||||
fullWidth
|
||||
|
|
|
@ -95,7 +95,7 @@ function UsersSearch() {
|
|||
>
|
||||
<FormattedMessage
|
||||
id="successfullyCreatedUser"
|
||||
defaultMessage="Successfully Updated User"
|
||||
defaultMessage="Successfully Created User"
|
||||
/>
|
||||
|
||||
<IconButton
|
||||
|
|
|
@ -215,7 +215,7 @@ function userForm(props: userFormProps) {
|
|||
<Box
|
||||
sx={{
|
||||
position: 'absolute',
|
||||
top: isMobile ? '50%' : '30%',
|
||||
top: isMobile ? '50%' : '40%',
|
||||
left: '50%',
|
||||
transform: 'translate(-50%, -50%)',
|
||||
width: 500,
|
||||
|
|
Loading…
Reference in New Issue