85 lines
2.4 KiB
C#
85 lines
2.4 KiB
C#
using System.Data;
|
|
using System.Text.Json;
|
|
using Confluent.Kafka;
|
|
using Dapper;
|
|
using GeoPulse_Pipeline;
|
|
using Microsoft.AspNetCore.Mvc;
|
|
using Npgsql;
|
|
|
|
var builder = WebApplication.CreateBuilder(args);
|
|
var kafkaHost = builder.Configuration.GetValue<string>("KafkaHost");
|
|
var producerConfig = new ProducerConfig()
|
|
{
|
|
BootstrapServers = kafkaHost,
|
|
Acks = Acks.Leader
|
|
};
|
|
|
|
var connString = builder.Configuration.GetConnectionString("DefaultConnection");
|
|
var kafkaProducer = new ProducerBuilder<Null, string>(producerConfig).Build();
|
|
builder.Services.AddSingleton(kafkaProducer);
|
|
builder.Services.AddEndpointsApiExplorer();
|
|
builder.Services.AddSwaggerGen();
|
|
builder.Services.AddHostedService<KafkaConsumer>();
|
|
builder.Services.AddNpgsqlDataSource(connString);
|
|
builder.Services.AddScoped<IDbConnection>(sp => sp.GetRequiredService<NpgsqlDataSource>().CreateConnection());
|
|
|
|
var app = builder.Build();
|
|
|
|
if (app.Environment.IsDevelopment())
|
|
{
|
|
app.UseSwagger();
|
|
app.UseSwaggerUI();
|
|
}
|
|
|
|
app.UseHttpsRedirection();
|
|
|
|
app.MapPost("/api/telemetry", async (
|
|
[FromBody] TelemetryRequest request,
|
|
[FromQuery] bool useKafka,
|
|
IProducer<Null, string> producer,
|
|
IDbConnection dbConnection) =>
|
|
{
|
|
if (string.IsNullOrWhiteSpace(request.DeviceId))
|
|
return Results.BadRequest("Device ID is required.");
|
|
|
|
if (useKafka)
|
|
{
|
|
var message = JsonSerializer.Serialize(request);
|
|
try
|
|
{
|
|
await producer.ProduceAsync("telemetry-events", new Message<Null, string> { Value = message });
|
|
return Results.Ok(new { Status = "Success", Route = "Kafka", Message = "已推入消息佇列" });
|
|
}
|
|
catch (ProduceException<Null, string> ex)
|
|
{
|
|
Console.WriteLine($"Kafka Delivery failed: {ex.Error.Reason}");
|
|
return Results.Problem("Kafka 寫入失敗");
|
|
}
|
|
}
|
|
else
|
|
{
|
|
try
|
|
{
|
|
string sql =
|
|
@"INSERT INTO telemetry_history (id, device_id, geom, timestamp) VALUES (gen_random_uuid(), @DeviceId, @Lng, @Lat, @Timestamp";
|
|
await dbConnection.ExecuteAsync(sql, request);
|
|
|
|
return Results.Ok(new { Status = "Success", Route = "Direct-DB", Message = "已直接寫入資料庫" });
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Console.WriteLine($"DB Insert failed: {ex.Message}");
|
|
return Results.Problem("資料庫寫入失敗");
|
|
}
|
|
}
|
|
});
|
|
|
|
app.Lifetime.ApplicationStopping.Register(() =>
|
|
{
|
|
kafkaProducer.Flush(TimeSpan.FromSeconds(10));
|
|
kafkaProducer.Dispose();
|
|
});
|
|
|
|
app.Run();
|
|
|
|
record TelemetryRequest(string DeviceId, double Lng, double Lat, DateTimeOffset Timestamp); |