|
|
@@ -13,12 +13,15 @@ import lombok.extern.slf4j.Slf4j;
|
|
|
import org.springframework.beans.factory.annotation.Autowired;
|
|
|
import org.springframework.beans.factory.annotation.Value;
|
|
|
import org.springframework.stereotype.Component;
|
|
|
+import java.io.ByteArrayInputStream;
|
|
|
import java.nio.charset.StandardCharsets;
|
|
|
-import java.nio.file.*;
|
|
|
import java.time.LocalDateTime;
|
|
|
import java.time.format.DateTimeFormatter;
|
|
|
import java.util.*;
|
|
|
import java.util.concurrent.*;
|
|
|
+import io.minio.MinioClient;
|
|
|
+import io.minio.PutObjectArgs;
|
|
|
+import io.minio.RemoveObjectArgs;
|
|
|
|
|
|
/** 只处理已组好的文本行或完整WAV,一份音频只计算和保存一次。 */
|
|
|
@Slf4j
|
|
|
@@ -27,7 +30,8 @@ import java.util.concurrent.*;
|
|
|
public class MessageHandler extends ChannelInboundHandlerAdapter {
|
|
|
public final ConcurrentMap<ChannelId, ChannelInfo> channelGroup = new ConcurrentHashMap<>();
|
|
|
public final ConcurrentMap<String, ChannelInfo> deviceGroup = new ConcurrentHashMap<>();
|
|
|
- @Value("${wlds.fileUploadBaseDir}") private String fileUploadBaseDir;
|
|
|
+ @Autowired private MinioClient minioClient;
|
|
|
+ @Value("${minio.bucket}") private String minioBucket;
|
|
|
// 正常/异常录音每24小时上报;连续重传按滑动静默窗口去重,下一份录音独立处理。
|
|
|
@Value("${wlds.retransmission-window-seconds:120}") private long retransmissionWindowSeconds = 120;
|
|
|
private final AudioUploadDeduplicator uploads = new AudioUploadDeduplicator();
|
|
|
@@ -81,37 +85,47 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
|
|
|
}
|
|
|
|
|
|
private void saveAudio(ChannelHandlerContext ctx, ChannelInfo info, byte[] bytes) throws Exception {
|
|
|
- Path file = null;
|
|
|
- boolean saved = false;
|
|
|
+ LocalDateTime now = LocalDateTime.now();
|
|
|
+ String dateDir = now.format(DateTimeFormatter.ofPattern("yyyy/MM/dd"));
|
|
|
+ String uuid = UUID.randomUUID().toString().replace("-", "");
|
|
|
+ // 统一归类到 MinIO 的 noise 业务目录,对象名:noise/{yyyy/MM/dd}/{uuid}.wav
|
|
|
+ String objectName = "noise/" + dateDir + "/" + uuid + ".wav";
|
|
|
+ NoiseInfo noise = new NoiseInfo();
|
|
|
+ boolean uploaded = false;
|
|
|
try {
|
|
|
- LocalDateTime now = LocalDateTime.now();
|
|
|
- String relative = "/devices/" + info.getDeviceNo() + "/" + now.toLocalDate().toString().replace("-", "")
|
|
|
- + "/" + UUID.randomUUID() + ".wav";
|
|
|
- Path base = Paths.get(fileUploadBaseDir).toAbsolutePath().normalize();
|
|
|
- file = base.resolve(relative.substring(1)).normalize();
|
|
|
- if (!file.startsWith(base)) throw new IllegalArgumentException("音频路径无效");
|
|
|
- Files.createDirectories(file.getParent());
|
|
|
- // 完整写入后再计算,避免异步追加造成不存在/不完整的文件。
|
|
|
- Files.write(file, bytes, StandardOpenOption.CREATE_NEW);
|
|
|
- NoiseInfo noise = new NoiseInfo();
|
|
|
+ // 先上传到 MinIO,成功后再入库,避免产生无法播放的孤立记录。
|
|
|
+ minioClient.putObject(PutObjectArgs.builder()
|
|
|
+ .bucket(minioBucket)
|
|
|
+ .object(objectName)
|
|
|
+ .stream(new ByteArrayInputStream(bytes), bytes.length, -1)
|
|
|
+ .contentType("audio/wav")
|
|
|
+ .build());
|
|
|
+ uploaded = true;
|
|
|
noise.setEncode(info.getDeviceNo());
|
|
|
noise.setPower(info.getPower());
|
|
|
noise.setGain(info.getGain());
|
|
|
- noise.setRms(DataParser.calculateRMS(file.toString()));
|
|
|
- noise.setCenterFrequency(DataParser.calcCenterFrequency(file.toString()));
|
|
|
- noise.setFilepath(relative);
|
|
|
+ noise.setRms(DataParser.calculateRMS(bytes));
|
|
|
+ noise.setCenterFrequency(DataParser.calcCenterFrequency(bytes));
|
|
|
+ String relativePath = "/" + objectName;
|
|
|
+ noise.setFilepath(relativePath);
|
|
|
noise.setTs(now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
|
|
|
noise.setCreateTime(new Date());
|
|
|
requireSaved(service.save(noise));
|
|
|
- saved = true;
|
|
|
- info.setPath(relative);
|
|
|
- info.setFilePath(relative);
|
|
|
+ info.setPath(relativePath);
|
|
|
+ info.setFilePath(relativePath);
|
|
|
acknowledge(ctx, noise.getEncode());
|
|
|
run(noise.getEncode(), "漏失报警预警", () -> noiseLeakAlarmService.evaluate(noise));
|
|
|
- } finally {
|
|
|
- if (!saved && file != null) {
|
|
|
- try { Files.deleteIfExists(file); } catch (Exception e) { log.warn("清理未入库音频失败", e); }
|
|
|
+ } catch (Exception e) {
|
|
|
+ // 入库失败则回滚已上传的 MinIO 对象,保持"只有成功入库的音频才保留"。
|
|
|
+ if (uploaded) {
|
|
|
+ try {
|
|
|
+ minioClient.removeObject(RemoveObjectArgs.builder()
|
|
|
+ .bucket(minioBucket).object(objectName).build());
|
|
|
+ } catch (Exception ignore) {
|
|
|
+ log.warn("清理上传失败的音频对象失败: {}", objectName, ignore);
|
|
|
+ }
|
|
|
}
|
|
|
+ throw e;
|
|
|
}
|
|
|
}
|
|
|
|