owner/RSE
1
1
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Fix NetMQ not stopping.

Browse Source
This commit is contained in:
owner
2024-11-25 19:13:37 +01:00
parent f3c6338a6a
commit 96b5ce4ff5
69 changed files with 120 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Backend/Handler/Communication.cs
+45 -40
View File
@@ -12,15 +12,14 @@ namespace Backend.Handler;
public class Communication
{
private readonly NetMQPoller _poller;
private readonly DbHandler _dbHandler;
private readonly ThreadHandler _threadHandler;
private bool _isRunning = true;
public Communication(DbHandler dbHandler, ThreadHandler threadHandler)
{
_dbHandler = dbHandler;
_threadHandler = threadHandler;
_poller = new();
}
public WaitHandle[] Start()
@@ -35,7 +34,7 @@ public class Communication
return waitHandles;
}
private void Server(object obj)
/*private void Server(object obj)
{
using ResponseSocket server = new();
server.Bind("tcp://*:5556");
@@ -50,17 +49,31 @@ public class Communication
Console.WriteLine("Communication stopped.");
((EventWaitHandle) obj).Set();
}*/
private void Server(object obj)
{
using ResponseSocket rep = new();
rep.Bind("tcp://127.0.0.1:5556");
while (_isRunning)
{
byte[] message = rep.ReceiveFrameBytes();
CommunicationObject communicationObject = MessagePackSerializer.Deserialize<CommunicationObject>(message);
OnServerOnReceiveReady(communicationObject, rep);
}
((EventWaitHandle) obj).Set();
}
[RequiresDynamicCode("Calls System.Text.Json.JsonSerializer.Serialize<TValue>(TValue, JsonSerializerOptions)")]
[RequiresUnreferencedCode("Calls System.Text.Json.JsonSerializer.Serialize<TValue>(TValue, JsonSerializerOptions)")]
private void OnServerOnReceiveReady(object? _, NetMQSocketEventArgs e)
private void OnServerOnReceiveReady(CommunicationObject communicationObject, ResponseSocket rep)
{
byte[] message = e.Socket.ReceiveFrameBytes();
CommunicationObject communicationObject = MessagePackSerializer.Deserialize<CommunicationObject>(message);
switch (communicationObject.Command)
{
case CommunicationCommand.GetScanningProgress:
@@ -68,10 +81,10 @@ public class Communication
DatabaseSizes databaseSizes = FilesystemHelper.GetDatabaseSizes();
long discardedIndexes = _dbHandler.GetDiscardedIndexes();
ScanningStatus status = new();
// 4294967296 is all Ipv4 addresses.
if (discardedIndexes != 0)
{
status.PercentageOfIpv4Scanned = (float)discardedIndexes / 4294967296 * 100;
@@ -80,67 +93,64 @@ public class Communication
{
status.PercentageOfIpv4Scanned = 0.0000000001f;
}
status.AmountOfIpv4Left = 4294967296 - discardedIndexes;
status.TotalFiltered = DbHandler.GetFilteredIndexes();
status.TotalDiscarded = discardedIndexes;
status.MyDbSize = databaseSizes.MyDbSize;
status.FilteredDbSize = databaseSizes.FilteredDbSize;
status.DiscardedDbSize = databaseSizes.DiscardedDbSize;
byte[] serializedResult = MessagePackSerializer.Serialize(status, MessagePackSerializerOptions.Standard.WithCompression(MessagePackCompression.Lz4BlockArray));
e.Socket.SendFrame(serializedResult);
rep.SendFrame(serializedResult);
break;
}
case CommunicationCommand.StopScanning:
SendStringResponse(e, "Server is stopping.");
_threadHandler.Stop();
break;
case CommunicationCommand.GarbageCollect:
ThreadHandler.ManualGc();
SendStringResponse(e, "Server has garbage collected.");
_threadHandler.Stop();
break;
case CommunicationCommand.DbReindex:
{
_dbHandler.ReIndex();
SendStringResponse(e, "All Dbs have been reindexed.");
SendStringResponse(rep, "All Dbs have been reindexed.");
break;
}
case CommunicationCommand.DbVacuum:
{
_dbHandler.Vacuum();
SendStringResponse(e, "All Dbs have been vacuumed.");
SendStringResponse(rep, "All Dbs have been vacuumed.");
break;
}
case CommunicationCommand.GetSearches:
{
if (!string.IsNullOrWhiteSpace(communicationObject.SearchTerm))
{
SendSearchResponse(e, communicationObject.SearchTerm);
SendSearchResponse(rep, communicationObject.SearchTerm);
}
break;
}
case CommunicationCommand.StopScanning:
{
_isRunning = false;
break;
}
}
}
private static void SendStringResponse(NetMQSocketEventArgs e, string response)
private static void SendStringResponse(ResponseSocket rep, string response)
{
MessagePackSerializerOptions withCompression = MessagePackSerializerOptions.Standard.WithCompression(MessagePackCompression.Lz4BlockArray);
byte[] serializedResult = MessagePackSerializer.Serialize(response, withCompression);
e.Socket.SendFrame(serializedResult);
rep.SendFrame(serializedResult);
}
private static void SendSearchResponse(NetMQSocketEventArgs e, string searchTerm)
private static void SendSearchResponse(ResponseSocket rep, string searchTerm)
{
//SearchResults result = SearchHelper.Search(communicationObject.SearchTerm!, _dbHandler);
SearchResults result = new()
@@ -157,11 +167,6 @@ public class Communication
string serializedResult = JsonSerializer.Serialize(result);
e.Socket.SendFrame(serializedResult);
}
public void Stop()
{
_poller.Stop();
rep.SendFrame(serializedResult);
}
}
Backend/Handler/ContentFilter.cs
+8 -3
View File
@@ -11,8 +11,8 @@ public class ContentFilter
{
private readonly ConcurrentQueue<QueueItem> _queue;
private readonly DbHandler _dbHandler;
private const string GetDomainPort80 = "/home/skingging/Documents/Projects/CSharp/RSE/Backend/GetDomainNamePort80.sh";
private const string GetDomainPort443 = "/home/skingging/Documents/Projects/CSharp/RSE/Backend/GetDomainNamePort443.sh";
private const string GetDomainPort80 = "/home/skingging/Documents/Projects/CSharp/RSE/Backend/Scripts/GetDomainNamePort80.sh";
private const string GetDomainPort443 = "/home/skingging/Documents/Projects/CSharp/RSE/Backend/Scripts/GetDomainNamePort443.sh";
private bool _stop;
public ContentFilter(ConcurrentQueue<QueueItem> queue, DbHandler dbHandler)
@@ -54,6 +54,11 @@ public class ContentFilter
_queue.Enqueue(superUnfilteredObject);
if (_dbHandler.GetFilteredIp(unfiltered.Ip))
{
continue;
}
Filtered filtered = GetSiteData(unfiltered.Ip);
filtered.Port1 = unfiltered.Port1;
@@ -230,7 +235,7 @@ public class ContentFilter
proc.Dispose();
}
}
public void Stop()
{
_stop = true;
Backend/Handler/ThreadHandler.cs
+10 -16
View File
@@ -29,20 +29,20 @@ public class ThreadHandler
public void Start()
{
//Thread scanner = new(StartScanner);
//Thread indexer = new(StartIndexer);
Thread scanner = new(StartScanner);
Thread indexer = new(StartIndexer);
Thread database = new(StartDbHandler);
Thread discarded = new(StartDiscardedDbHandler);
Thread communication = new(StartCommunicationHandler);
//scanner.Start();
//indexer.Start();
scanner.Start();
indexer.Start();
database.Start();
discarded.Start();
communication.Start();
//scanner.Join();
//indexer.Join();
scanner.Join();
indexer.Join();
database.Join();
discarded.Join();
communication.Join();
@@ -107,21 +107,15 @@ public class ThreadHandler
Console.WriteLine("Communicator finished");
_communicationStopped = true;
Stop();
}
private void StopCommunicator()
{
Thread t = new(_communication.Stop);
t.Start();
t.Join();
}
public void Stop()
private void Stop()
{
_stopSignal = true;
_ipScanner.Stop();
_contentFilter.Stop();
StopCommunicator();
bool stopping = true;