501 lines
17 KiB
C#
501 lines
17 KiB
C#
using System.Net;
|
|
using System.Net.Sockets;
|
|
using System.Collections.Concurrent;
|
|
using System.Security.Cryptography;
|
|
using System.Text;
|
|
using System.Text.Json;
|
|
using MessagePack;
|
|
using Mimir.Runtime.Synchronization;
|
|
|
|
var options = EveSensorReceiverOptions.Parse(args);
|
|
using var receiver = new EveSensorReceiver(options);
|
|
await receiver.RunAsync();
|
|
|
|
internal sealed record EveSensorReceiverOptions(
|
|
int Port,
|
|
string Path,
|
|
string SourceId,
|
|
string ExpectedType,
|
|
string SubscribePath)
|
|
{
|
|
public static EveSensorReceiverOptions Parse(string[] args)
|
|
{
|
|
return new EveSensorReceiverOptions(
|
|
ParseInt(args, "--port", 8793),
|
|
ParseString(args, "--path", "/eve/camera"),
|
|
ParseString(args, "--source-id", "eve-camera"),
|
|
ParseString(args, "--type", "video-frame"),
|
|
ParseString(args, "--subscribe-path", ParseString(args, "--path", "/eve/camera") + "/subscribe"));
|
|
}
|
|
|
|
private static string ParseString(IReadOnlyList<string> args, string name, string fallback)
|
|
{
|
|
for (var index = 0; index < args.Count - 1; index++)
|
|
{
|
|
if (string.Equals(args[index], name, StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
return args[index + 1];
|
|
}
|
|
}
|
|
|
|
return fallback;
|
|
}
|
|
|
|
private static int ParseInt(IReadOnlyList<string> args, string name, int fallback)
|
|
{
|
|
return int.TryParse(ParseString(args, name, ""), out var value) ? value : fallback;
|
|
}
|
|
}
|
|
|
|
internal sealed class EveSensorReceiver(EveSensorReceiverOptions options) : IDisposable
|
|
{
|
|
private readonly TcpListener listener = new(IPAddress.Any, options.Port);
|
|
private readonly CancellationTokenSource stopping = new();
|
|
private readonly ConcurrentDictionary<int, WebSocketSubscriber> subscribers = new();
|
|
private int nextSubscriberId;
|
|
|
|
public async Task RunAsync()
|
|
{
|
|
listener.Start();
|
|
Console.Error.WriteLine($"Eve sensor receiver listening on ws://0.0.0.0:{options.Port}{options.Path} source={options.SourceId}");
|
|
while (!stopping.IsCancellationRequested)
|
|
{
|
|
var client = await listener.AcceptTcpClientAsync(stopping.Token).ConfigureAwait(false);
|
|
_ = Task.Run(() => HandleAsync(client));
|
|
}
|
|
}
|
|
|
|
private async Task HandleAsync(TcpClient tcpClient)
|
|
{
|
|
await using var stream = tcpClient.GetStream();
|
|
var request = await ReadHttpRequestAsync(stream).ConfigureAwait(false);
|
|
if (request.Path == "/health")
|
|
{
|
|
var health = JsonSerializer.Serialize(new
|
|
{
|
|
ok = true,
|
|
options.SourceId,
|
|
options.ExpectedType,
|
|
options.Path,
|
|
options.SubscribePath,
|
|
Subscribers = subscribers.Count,
|
|
});
|
|
await WriteHttpResponseAsync(stream, "200 OK", "application/json", Encoding.UTF8.GetBytes(health)).ConfigureAwait(false);
|
|
return;
|
|
}
|
|
|
|
if (string.Equals(request.Path, options.SubscribePath, StringComparison.Ordinal)
|
|
&& request.Headers.TryGetValue("Sec-WebSocket-Key", out var subscriberKey))
|
|
{
|
|
await WriteWebSocketHandshakeAsync(stream, subscriberKey).ConfigureAwait(false);
|
|
await SubscribeAsync(stream).ConfigureAwait(false);
|
|
return;
|
|
}
|
|
|
|
if (!string.Equals(request.Path, options.Path, StringComparison.Ordinal)
|
|
|| !request.Headers.TryGetValue("Sec-WebSocket-Key", out var key))
|
|
{
|
|
await WriteHttpResponseAsync(stream, "404 Not Found", "text/plain", Encoding.UTF8.GetBytes("not found")).ConfigureAwait(false);
|
|
return;
|
|
}
|
|
|
|
await WriteWebSocketHandshakeAsync(stream, key).ConfigureAwait(false);
|
|
while (!stopping.IsCancellationRequested)
|
|
{
|
|
var frame = await ReceiveFrameAsync(stream).ConfigureAwait(false);
|
|
if (frame.Opcode == 0x8)
|
|
{
|
|
await WriteCloseFrameAsync(stream).ConfigureAwait(false);
|
|
break;
|
|
}
|
|
|
|
if (frame.Opcode == 0x2)
|
|
{
|
|
if (TryValidateCultMeshObservation(frame.Payload, out var normalizedObservation)
|
|
|| TryValidateCultMeshMediaObservation(frame.Payload, out normalizedObservation))
|
|
{
|
|
Console.WriteLine(normalizedObservation);
|
|
Console.Out.Flush();
|
|
await BroadcastAsync(normalizedObservation).ConfigureAwait(false);
|
|
}
|
|
|
|
continue;
|
|
}
|
|
|
|
if (frame.Opcode != 0x1)
|
|
{
|
|
continue;
|
|
}
|
|
|
|
var text = Encoding.UTF8.GetString(frame.Payload);
|
|
if (TryValidateFrameEvent(text, out var normalized))
|
|
{
|
|
Console.WriteLine(normalized);
|
|
Console.Out.Flush();
|
|
await BroadcastAsync(normalized).ConfigureAwait(false);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task SubscribeAsync(NetworkStream stream)
|
|
{
|
|
var id = Interlocked.Increment(ref nextSubscriberId);
|
|
var subscriber = new WebSocketSubscriber(id, stream);
|
|
subscribers[id] = subscriber;
|
|
Console.Error.WriteLine($"Eve sensor receiver subscriber {id} connected on {options.SubscribePath}.");
|
|
try
|
|
{
|
|
await subscriber.SendTextAsync(JsonSerializer.Serialize(new
|
|
{
|
|
type = "mimir-verse-subscription",
|
|
status = "connected",
|
|
options.Path,
|
|
options.SubscribePath,
|
|
options.SourceId,
|
|
options.ExpectedType,
|
|
})).ConfigureAwait(false);
|
|
|
|
while (!stopping.IsCancellationRequested)
|
|
{
|
|
var frame = await ReceiveFrameAsync(stream).ConfigureAwait(false);
|
|
if (frame.Opcode == 0x8)
|
|
{
|
|
await WriteCloseFrameAsync(stream).ConfigureAwait(false);
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
catch (IOException)
|
|
{
|
|
}
|
|
catch (SocketException)
|
|
{
|
|
}
|
|
finally
|
|
{
|
|
subscribers.TryRemove(id, out _);
|
|
Console.Error.WriteLine($"Eve sensor receiver subscriber {id} disconnected.");
|
|
}
|
|
}
|
|
|
|
private async Task BroadcastAsync(string normalized)
|
|
{
|
|
foreach (var pair in subscribers.ToArray())
|
|
{
|
|
try
|
|
{
|
|
await pair.Value.SendTextAsync(normalized).ConfigureAwait(false);
|
|
}
|
|
catch (IOException)
|
|
{
|
|
subscribers.TryRemove(pair.Key, out _);
|
|
}
|
|
catch (SocketException)
|
|
{
|
|
subscribers.TryRemove(pair.Key, out _);
|
|
}
|
|
catch (ObjectDisposedException)
|
|
{
|
|
subscribers.TryRemove(pair.Key, out _);
|
|
}
|
|
}
|
|
}
|
|
|
|
private bool TryValidateFrameEvent(string text, out string normalized)
|
|
{
|
|
normalized = "";
|
|
try
|
|
{
|
|
using var document = JsonDocument.Parse(text);
|
|
var root = document.RootElement;
|
|
if (!root.TryGetProperty("type", out var type)
|
|
|| !string.Equals(type.GetString(), options.ExpectedType, StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
if (!string.Equals(options.ExpectedType, "cultmesh-observation", StringComparison.OrdinalIgnoreCase)
|
|
&& (!root.TryGetProperty("sourceId", out var sourceId)
|
|
|| !string.Equals(sourceId.GetString(), options.SourceId, StringComparison.Ordinal)))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
normalized = root.GetRawText();
|
|
return true;
|
|
}
|
|
catch (JsonException)
|
|
{
|
|
return false;
|
|
}
|
|
}
|
|
|
|
private bool TryValidateCultMeshObservation(byte[] payload, out string normalized)
|
|
{
|
|
normalized = "";
|
|
try
|
|
{
|
|
var observation = MessagePackSerializer.Deserialize<MimirEveSensorObservationDocument>(payload);
|
|
if (!string.Equals(observation.StreamId, options.SourceId, StringComparison.Ordinal) &&
|
|
!string.Equals(options.ExpectedType, "cultmesh-observation", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
normalized = JsonSerializer.Serialize(new
|
|
{
|
|
type = "cultmesh-observation",
|
|
document = "mimir.eve_sensor_observation.v1",
|
|
observation.ObservationId,
|
|
observation.DeviceId,
|
|
observation.StreamId,
|
|
observation.Kind,
|
|
observation.Sequence,
|
|
observation.SensorTimestampNs,
|
|
observation.ElapsedRealtimeNs,
|
|
observation.WallClockUtc,
|
|
observation.ClockDomainId,
|
|
observation.Values,
|
|
observation.Action,
|
|
observation.PointerCount,
|
|
observation.X,
|
|
observation.Y,
|
|
observation.Accuracy,
|
|
});
|
|
return true;
|
|
}
|
|
catch
|
|
{
|
|
return false;
|
|
}
|
|
}
|
|
|
|
private bool TryValidateCultMeshMediaObservation(byte[] payload, out string normalized)
|
|
{
|
|
normalized = "";
|
|
try
|
|
{
|
|
var observation = MessagePackSerializer.Deserialize<MimirEveMediaObservationDocument>(payload);
|
|
if (!string.Equals(observation.StreamId, options.SourceId, StringComparison.Ordinal) &&
|
|
!string.Equals(options.ExpectedType, "cultmesh-observation", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
normalized = JsonSerializer.Serialize(new
|
|
{
|
|
type = "cultmesh-media-observation",
|
|
document = "mimir.eve_media_observation.v1",
|
|
observation.ObservationId,
|
|
observation.DeviceId,
|
|
observation.StreamId,
|
|
observation.Kind,
|
|
observation.Sequence,
|
|
observation.SensorTimestampNs,
|
|
observation.ElapsedRealtimeNs,
|
|
observation.WallClockUtc,
|
|
observation.ClockDomainId,
|
|
observation.Format,
|
|
observation.Width,
|
|
observation.Height,
|
|
observation.SampleRate,
|
|
observation.Channels,
|
|
observation.FrameCount,
|
|
observation.PayloadEncoding,
|
|
PayloadBytes = observation.Payload.Length,
|
|
});
|
|
return true;
|
|
}
|
|
catch
|
|
{
|
|
return false;
|
|
}
|
|
}
|
|
|
|
private static async Task<HttpRequest> ReadHttpRequestAsync(NetworkStream stream)
|
|
{
|
|
var bytes = new List<byte>();
|
|
var lastFour = new Queue<byte>(4);
|
|
while (true)
|
|
{
|
|
var value = stream.ReadByte();
|
|
if (value < 0)
|
|
{
|
|
throw new IOException("Client closed before HTTP headers completed.");
|
|
}
|
|
|
|
var b = (byte)value;
|
|
bytes.Add(b);
|
|
lastFour.Enqueue(b);
|
|
if (lastFour.Count > 4)
|
|
{
|
|
lastFour.Dequeue();
|
|
}
|
|
|
|
if (lastFour.Count == 4 && lastFour.SequenceEqual(new byte[] { 13, 10, 13, 10 }))
|
|
{
|
|
break;
|
|
}
|
|
}
|
|
|
|
var text = Encoding.ASCII.GetString(bytes.ToArray());
|
|
var lines = text.Split("\r\n", StringSplitOptions.None);
|
|
var first = lines[0].Split(' ');
|
|
var headers = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
|
|
foreach (var line in lines.Skip(1))
|
|
{
|
|
var colon = line.IndexOf(':');
|
|
if (colon > 0)
|
|
{
|
|
headers[line[..colon].Trim()] = line[(colon + 1)..].Trim();
|
|
}
|
|
}
|
|
|
|
await Task.CompletedTask.ConfigureAwait(false);
|
|
return new HttpRequest(first.Length > 1 ? first[1] : "/", headers);
|
|
}
|
|
|
|
private static async Task WriteHttpResponseAsync(NetworkStream stream, string status, string contentType, byte[] body)
|
|
{
|
|
var header = Encoding.ASCII.GetBytes(
|
|
$"HTTP/1.1 {status}\r\nContent-Type: {contentType}\r\nContent-Length: {body.Length}\r\nConnection: close\r\n\r\n");
|
|
await stream.WriteAsync(header).ConfigureAwait(false);
|
|
await stream.WriteAsync(body).ConfigureAwait(false);
|
|
}
|
|
|
|
private static async Task WriteWebSocketHandshakeAsync(NetworkStream stream, string key)
|
|
{
|
|
var acceptBytes = SHA1.HashData(Encoding.ASCII.GetBytes(key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"));
|
|
var accept = Convert.ToBase64String(acceptBytes);
|
|
var response = Encoding.ASCII.GetBytes(
|
|
"HTTP/1.1 101 Switching Protocols\r\n" +
|
|
"Upgrade: websocket\r\n" +
|
|
"Connection: Upgrade\r\n" +
|
|
$"Sec-WebSocket-Accept: {accept}\r\n\r\n");
|
|
await stream.WriteAsync(response).ConfigureAwait(false);
|
|
}
|
|
|
|
private static async Task<WebSocketFrame> ReceiveFrameAsync(NetworkStream stream)
|
|
{
|
|
var header = new byte[2];
|
|
await ReadExactAsync(stream, header).ConfigureAwait(false);
|
|
var opcode = header[0] & 0x0f;
|
|
var masked = (header[1] & 0x80) != 0;
|
|
var length = header[1] & 0x7f;
|
|
if (length == 126)
|
|
{
|
|
var extended = new byte[2];
|
|
await ReadExactAsync(stream, extended).ConfigureAwait(false);
|
|
length = (extended[0] << 8) | extended[1];
|
|
}
|
|
else if (length == 127)
|
|
{
|
|
var extended = new byte[8];
|
|
await ReadExactAsync(stream, extended).ConfigureAwait(false);
|
|
var longLength = 0L;
|
|
foreach (var b in extended)
|
|
{
|
|
longLength = (longLength << 8) | b;
|
|
}
|
|
|
|
if (longLength > int.MaxValue)
|
|
{
|
|
throw new InvalidDataException("WebSocket frame is too large.");
|
|
}
|
|
|
|
length = (int)longLength;
|
|
}
|
|
|
|
var mask = new byte[4];
|
|
if (masked)
|
|
{
|
|
await ReadExactAsync(stream, mask).ConfigureAwait(false);
|
|
}
|
|
|
|
var payload = new byte[length];
|
|
await ReadExactAsync(stream, payload).ConfigureAwait(false);
|
|
if (masked)
|
|
{
|
|
for (var index = 0; index < payload.Length; index++)
|
|
{
|
|
payload[index] ^= mask[index % 4];
|
|
}
|
|
}
|
|
|
|
return new WebSocketFrame(opcode, payload);
|
|
}
|
|
|
|
private static Task WriteCloseFrameAsync(NetworkStream stream)
|
|
{
|
|
return stream.WriteAsync(new byte[] { 0x88, 0x00 }).AsTask();
|
|
}
|
|
|
|
private static async Task ReadExactAsync(NetworkStream stream, byte[] buffer)
|
|
{
|
|
var offset = 0;
|
|
while (offset < buffer.Length)
|
|
{
|
|
var read = await stream.ReadAsync(buffer.AsMemory(offset, buffer.Length - offset)).ConfigureAwait(false);
|
|
if (read == 0)
|
|
{
|
|
throw new IOException("WebSocket closed.");
|
|
}
|
|
|
|
offset += read;
|
|
}
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
stopping.Cancel();
|
|
listener.Stop();
|
|
stopping.Dispose();
|
|
}
|
|
}
|
|
|
|
internal sealed record HttpRequest(string Path, IReadOnlyDictionary<string, string> Headers);
|
|
|
|
internal sealed record WebSocketFrame(int Opcode, byte[] Payload);
|
|
|
|
internal sealed class WebSocketSubscriber(int id, NetworkStream stream)
|
|
{
|
|
private readonly SemaphoreSlim writeLock = new(1, 1);
|
|
|
|
public int Id => id;
|
|
|
|
public async Task SendTextAsync(string text)
|
|
{
|
|
var payload = Encoding.UTF8.GetBytes(text);
|
|
var header = new List<byte> { 0x81 };
|
|
if (payload.Length < 126)
|
|
{
|
|
header.Add((byte)payload.Length);
|
|
}
|
|
else if (payload.Length <= ushort.MaxValue)
|
|
{
|
|
header.Add(126);
|
|
header.Add((byte)(payload.Length >> 8));
|
|
header.Add((byte)(payload.Length & 0xff));
|
|
}
|
|
else
|
|
{
|
|
header.Add(127);
|
|
for (var shift = 56; shift >= 0; shift -= 8)
|
|
{
|
|
header.Add((byte)(((ulong)payload.Length >> shift) & 0xff));
|
|
}
|
|
}
|
|
|
|
await writeLock.WaitAsync().ConfigureAwait(false);
|
|
try
|
|
{
|
|
await stream.WriteAsync(header.ToArray()).ConfigureAwait(false);
|
|
await stream.WriteAsync(payload).ConfigureAwait(false);
|
|
await stream.FlushAsync().ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
writeLock.Release();
|
|
}
|
|
}
|
|
}
|