feat(audio): 添加机器人实时语音对讲功能
- 在GrpcServiceManager中注册机器人实时语音双向流服务异步Stub - 引入webrtc-java依赖支持WebRTC功能 - 新增AudioService实现WebSocket与gRPC音频流转换业务逻辑 - 实现WebSocket信令处理和音频数据传输功能 - 添加机器人音频心跳检测和会话管理机制 - 新增RobotAudioGrpcAdapter实现gRPC双向流音频传输 - 定义RobotAudioService gRPC服务协议和消息格式 - 实现音频帧队列管理和流量控制机制
This commit is contained in:
parent
710f33d882
commit
885e94883b
@ -0,0 +1,123 @@
|
||||
package com.cmvr.web.controller.api;
|
||||
|
||||
import com.cmvr.edge.client.service.AudioService;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.databind.node.ObjectNode;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.web.socket.BinaryMessage;
|
||||
import org.springframework.web.socket.CloseStatus;
|
||||
import org.springframework.web.socket.TextMessage;
|
||||
import org.springframework.web.socket.WebSocketSession;
|
||||
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
|
||||
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
|
||||
import org.springframework.web.socket.handler.AbstractWebSocketHandler;
|
||||
|
||||
/**
|
||||
* WebRTC 信令 WebSocket 入口。
|
||||
*
|
||||
* 路径：/ws/signaling
|
||||
*
|
||||
* 约束：
|
||||
* 1. WebSocket 只传 JSON 信令：join、offer、answer、candidate、heartbeat、stop；
|
||||
* 2. WebSocket 禁止传输 PCM 音频；
|
||||
* 3. PCM 音频由 WebRTC 媒体通道进入 Java 后端，再由 AudioService 转 gRPC 双向流。
|
||||
*/
|
||||
@Configuration
|
||||
@RequiredArgsConstructor
|
||||
public class WebRtcSignalingController implements WebSocketConfigurer {
|
||||
|
||||
private final AudioService audioService;
|
||||
|
||||
@Override
|
||||
public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
|
||||
registry.addHandler(new SignalingWebSocketHandler(audioService), "/ws/signaling")
|
||||
.setAllowedOrigins("*");
|
||||
registry.addHandler(new SignalingWebSocketHandler(audioService), "/ws/audio")
|
||||
.setAllowedOrigins("*");
|
||||
}
|
||||
|
||||
/**
|
||||
* 信令处理器。这里只处理文本 JSON，不处理二进制音频。
|
||||
*/
|
||||
private static class SignalingWebSocketHandler extends AbstractWebSocketHandler {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(SignalingWebSocketHandler.class);
|
||||
|
||||
private final AudioService audioService;
|
||||
private final ObjectMapper objectMapper = new ObjectMapper();
|
||||
|
||||
private SignalingWebSocketHandler(AudioService audioService) {
|
||||
this.audioService = audioService;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterConnectionEstablished(@NotNull WebSocketSession session) {
|
||||
log.info("WebRTC信令连接建立，wsSessionId={}", session.getId());
|
||||
sendConnected(session);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void handleTextMessage(@NotNull WebSocketSession session, @NotNull TextMessage message) {
|
||||
try {
|
||||
audioService.handleSignalingMessage(session, message.getPayload());
|
||||
} catch (Exception e) {
|
||||
log.error("处理WebRTC信令异常，wsSessionId={}", session.getId(), e);
|
||||
sendError(session, "处理WebRTC信令异常：" + e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void handleBinaryMessage(@NotNull WebSocketSession session, @NotNull BinaryMessage message) {
|
||||
try {
|
||||
audioService.handleBrowserAudioMessage(session, message.getPayload());
|
||||
} catch (Exception e) {
|
||||
log.error("处理WebSocket音频二进制帧失败，wsSessionId={}", session.getId(), e);
|
||||
sendError(session, "audio binary frame failed: " + e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void handleTransportError(@NotNull WebSocketSession session, @NotNull Throwable exception) {
|
||||
log.error("WebRTC信令连接传输异常，wsSessionId={}", session.getId(), exception);
|
||||
audioService.handleWebSocketClosed(session);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterConnectionClosed(@NotNull WebSocketSession session, @NotNull CloseStatus status) {
|
||||
log.info("WebRTC信令连接关闭，wsSessionId={}，status={}", session.getId(), status);
|
||||
audioService.handleWebSocketClosed(session);
|
||||
}
|
||||
|
||||
private void sendConnected(WebSocketSession session) {
|
||||
ObjectNode node = objectMapper.createObjectNode();
|
||||
node.put("type", "connected");
|
||||
node.put("timestamp", System.currentTimeMillis());
|
||||
sendJson(session, node);
|
||||
}
|
||||
|
||||
private void sendError(WebSocketSession session, String message) {
|
||||
ObjectNode node = objectMapper.createObjectNode();
|
||||
node.put("type", "error");
|
||||
node.put("message", message);
|
||||
node.put("timestamp", System.currentTimeMillis());
|
||||
sendJson(session, node);
|
||||
}
|
||||
|
||||
private void sendJson(WebSocketSession session, ObjectNode node) {
|
||||
if (session == null || !session.isOpen()) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
synchronized (session) {
|
||||
session.sendMessage(new TextMessage(objectMapper.writeValueAsString(node)));
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("发送WebRTC信令响应失败，wsSessionId={}", session.getId(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -55,6 +55,12 @@
|
||||
<version>1.5.9</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>dev.onvoid.webrtc</groupId>
|
||||
<artifactId>webrtc-java</artifactId>
|
||||
<version>0.14.0</version>
|
||||
</dependency>
|
||||
|
||||
|
||||
</dependencies>
|
||||
</project>
|
||||
</project>
|
||||
|
||||
@ -0,0 +1,743 @@
|
||||
package com.cmvr.edge.client.adapter;
|
||||
|
||||
import cmvr.api.RobotAudio;
|
||||
import cmvr.api.RobotAudioServiceGrpc;
|
||||
import com.google.protobuf.ByteString;
|
||||
import com.cmvr.edge.client.manage.GrpcServiceManager;
|
||||
import com.cmvr.edge.client.utils.EdgeCommonUtil;
|
||||
import io.grpc.Status;
|
||||
import io.grpc.stub.StreamObserver;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.OutputStream;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
import java.nio.file.StandardOpenOption;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
/**
|
||||
* 机器人实时语音 gRPC 适配层。
|
||||
*
|
||||
* 约束：
|
||||
* 1. Java 后端只作为 gRPC client；
|
||||
* 2. 机器人终端作为 gRPC server；
|
||||
* 3. Stub 必须复用项目现有 GrpcServiceManager，禁止在这里创建 ManagedChannel。
|
||||
*/
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class RobotAudioGrpcAdapter {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(RobotAudioGrpcAdapter.class);
|
||||
|
||||
/** 浏览器与机器人之间约定的音频格式：PCM-48K-16bit-单声道。 */
|
||||
public static final int SAMPLE_RATE = 48000;
|
||||
public static final int CHANNELS = 1;
|
||||
public static final int BITS_PER_SAMPLE = 16;
|
||||
// WebRTC 与主流 PCM 设备默认使用 signed 16-bit little-endian，终端如需大端序应在设备边界自行转换。
|
||||
public static final String ENCODING = "PCM_S16LE";
|
||||
private static final int FRAME_MS = 10;
|
||||
private static final int FRAME_SAMPLES = SAMPLE_RATE * FRAME_MS / 1000;
|
||||
private static final int FRAME_BYTES = FRAME_SAMPLES * CHANNELS * (BITS_PER_SAMPLE / 8);
|
||||
private static final int TARGET_QUEUED_BROWSER_FRAMES = 2;
|
||||
private static final int MAX_QUEUED_BROWSER_FRAMES = 10;
|
||||
private static final long AUDIO_STATS_LOG_INTERVAL_MS = 5000L;
|
||||
private static final int DEBUG_DUMP_SECONDS = 10;
|
||||
private static final int DEBUG_DUMP_BYTES = SAMPLE_RATE * CHANNELS * (BITS_PER_SAMPLE / 8) * DEBUG_DUMP_SECONDS;
|
||||
|
||||
private final GrpcServiceManager grpcServiceManager;
|
||||
private final ConcurrentHashMap<String, Long> lastGrpcInboundFormatLogAt = new ConcurrentHashMap<>();
|
||||
private final WavDumpRecorder robotInboundDumpRecorder =
|
||||
new WavDumpRecorder("robot-to-backend", DEBUG_DUMP_BYTES);
|
||||
|
||||
/**
|
||||
* 打开机器人音频双向流。
|
||||
*/
|
||||
public RobotAudioStream openStream(String terminalId,
|
||||
String deviceId,
|
||||
String sessionId,
|
||||
String operatorId,
|
||||
RobotAudioStreamListener listener) {
|
||||
RobotAudioServiceGrpc.RobotAudioServiceStub stub =
|
||||
grpcServiceManager.getGrpcClient(terminalId, RobotAudioServiceGrpc.RobotAudioServiceStub.class);
|
||||
|
||||
StreamObserver<RobotAudio.RobotAudioResponse> responseObserver = new StreamObserver<RobotAudio.RobotAudioResponse>() {
|
||||
@Override
|
||||
public void onNext(RobotAudio.RobotAudioResponse response) {
|
||||
try {
|
||||
handleRobotResponse(sessionId, response, listener);
|
||||
} catch (Exception e) {
|
||||
log.error("处理机器人音频响应失败，sessionId={}", sessionId, e);
|
||||
listener.onError(sessionId, e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable t) {
|
||||
log.error("机器人音频gRPC流异常，sessionId={}", sessionId, t);
|
||||
listener.onError(sessionId, t);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
log.info("机器人音频gRPC流结束，sessionId={}", sessionId);
|
||||
listener.onCompleted(sessionId);
|
||||
}
|
||||
};
|
||||
|
||||
StreamObserver<RobotAudio.RobotAudioRequest> requestObserver = stub.audioTalk(responseObserver);
|
||||
RobotAudioStream stream = new RobotAudioStream(terminalId, deviceId, sessionId, requestObserver);
|
||||
stream.sendStart(operatorId);
|
||||
return stream;
|
||||
}
|
||||
|
||||
private void handleRobotResponse(String sessionId,
|
||||
RobotAudio.RobotAudioResponse response,
|
||||
RobotAudioStreamListener listener) {
|
||||
if (response.hasHeader()
|
||||
&& !response.getHeader().getSuccess()
|
||||
&& response.getHeader().getErrorMessage() != null
|
||||
&& !response.getHeader().getErrorMessage().trim().isEmpty()) {
|
||||
String errorMessage = response.getHeader().getErrorMessage();
|
||||
log.warn("机器人音频响应失败，sessionId={}，error={}", sessionId, errorMessage);
|
||||
listener.onError(sessionId, new RuntimeException(errorMessage));
|
||||
return;
|
||||
}
|
||||
|
||||
switch (response.getPayloadCase()) {
|
||||
case START_ACK:
|
||||
listener.onStarted(sessionId, response.getStartAck().getAccepted(), response.getStartAck().getMessage());
|
||||
break;
|
||||
case AUDIO_FRAME:
|
||||
RobotAudio.RobotAudioFrame frame = response.getAudioFrame();
|
||||
byte[] pcm = frame.getPcm().toByteArray();
|
||||
robotInboundDumpRecorder.record(pcm);
|
||||
logGrpcInboundFormatIfNeeded(sessionId, frame);
|
||||
listener.onRobotAudio(sessionId, frame.getSequence(), frame.getTimestampMs(), pcm);
|
||||
break;
|
||||
case HEARTBEAT:
|
||||
listener.onHeartbeat(sessionId, response.getHeartbeat().getTimestampMs());
|
||||
break;
|
||||
case STOP:
|
||||
listener.onRobotStop(sessionId, response.getStop().getReason());
|
||||
break;
|
||||
case ERROR:
|
||||
RobotAudio.RobotAudioError error = response.getError();
|
||||
listener.onError(sessionId, new RuntimeException(error.getCode() + ":" + error.getMessage()));
|
||||
break;
|
||||
case PAYLOAD_NOT_SET:
|
||||
default:
|
||||
log.debug("机器人音频响应为空payload，sessionId={}", sessionId);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private void logGrpcInboundFormatIfNeeded(String sessionId, RobotAudio.RobotAudioFrame frame) {
|
||||
long now = System.currentTimeMillis();
|
||||
Long last = lastGrpcInboundFormatLogAt.get(sessionId);
|
||||
if (last != null && now - last < AUDIO_STATS_LOG_INTERVAL_MS) {
|
||||
return;
|
||||
}
|
||||
lastGrpcInboundFormatLogAt.put(sessionId, now);
|
||||
int bytes = frame.getPcm().size();
|
||||
int frameSamples = bytes / (CHANNELS * (BITS_PER_SAMPLE / 8));
|
||||
int frameMs = frameSamples * 1000 / SAMPLE_RATE;
|
||||
log.info("gRPC机器人上行音频格式，sessionId={}，sampleRate={}，channels={}，bitsPerSample={}，encoding={}，frameSamples={}，frameMs={}，bytes={}，sequence={}，timestampMs={}",
|
||||
sessionId,
|
||||
SAMPLE_RATE,
|
||||
CHANNELS,
|
||||
BITS_PER_SAMPLE,
|
||||
ENCODING,
|
||||
frameSamples,
|
||||
frameMs,
|
||||
bytes,
|
||||
frame.getSequence(),
|
||||
frame.getTimestampMs());
|
||||
logPcmStats("grpc-inbound-robot", sessionId, frame.getPcm().toByteArray(), frame.getSequence(), frame.getTimestampMs());
|
||||
}
|
||||
|
||||
private void logPcmStats(String direction, String sessionId, byte[] pcm, long sequence, long timestampMs) {
|
||||
PcmStats stats = PcmStats.fromPcm16Le(pcm);
|
||||
log.info("PCM-STATS direction={} sessionId={} sequence={} timestampMs={} bytes={} samples={} rms={} peak={} zeroCrossRate={} firstSamples={}",
|
||||
direction,
|
||||
sessionId,
|
||||
sequence,
|
||||
timestampMs,
|
||||
pcm == null ? 0 : pcm.length,
|
||||
stats.samples,
|
||||
stats.rms,
|
||||
stats.peak,
|
||||
stats.zeroCrossRate,
|
||||
stats.firstSamples);
|
||||
}
|
||||
|
||||
/**
|
||||
* 单个会话对应的一条 gRPC 双向流。
|
||||
*/
|
||||
public class RobotAudioStream {
|
||||
private final String terminalId;
|
||||
private final String deviceId;
|
||||
private final String sessionId;
|
||||
private final StreamObserver<RobotAudio.RobotAudioRequest> requestObserver;
|
||||
private final AtomicLong sequence = new AtomicLong(0);
|
||||
private final AtomicBoolean closed = new AtomicBoolean(false);
|
||||
private final Object sendLock = new Object();
|
||||
private final ConcurrentLinkedQueue<byte[]> browserAudioQueue = new ConcurrentLinkedQueue<>();
|
||||
private final AtomicInteger queuedBrowserFrames = new AtomicInteger(0);
|
||||
private final Object browserAudioBufferLock = new Object();
|
||||
private byte[] browserAudioRemainder = new byte[FRAME_BYTES];
|
||||
private int browserAudioRemainderLength;
|
||||
private final ScheduledExecutorService browserAudioSender;
|
||||
private final ScheduledFuture<?> browserAudioSenderFuture;
|
||||
private volatile long browserAudioFramesEnqueued;
|
||||
private volatile long browserAudioFramesSent;
|
||||
private volatile long browserAudioFramesDropped;
|
||||
private volatile long browserAudioSilentFramesSent;
|
||||
private volatile boolean browserAudioStarted;
|
||||
private volatile long lastAudioStatsLogAt;
|
||||
private volatile long lastGrpcOutboundFormatLogAt;
|
||||
|
||||
private RobotAudioStream(String terminalId,
|
||||
String deviceId,
|
||||
String sessionId,
|
||||
StreamObserver<RobotAudio.RobotAudioRequest> requestObserver) {
|
||||
this.terminalId = terminalId;
|
||||
this.deviceId = deviceId;
|
||||
this.sessionId = sessionId;
|
||||
this.requestObserver = requestObserver;
|
||||
this.browserAudioSender = Executors.newSingleThreadScheduledExecutor(runnable -> {
|
||||
Thread thread = new Thread(runnable, "robot-audio-send-" + sessionId);
|
||||
thread.setDaemon(true);
|
||||
return thread;
|
||||
});
|
||||
this.browserAudioSenderFuture = browserAudioSender.scheduleAtFixedRate(
|
||||
this::sendQueuedBrowserAudio,
|
||||
FRAME_MS,
|
||||
FRAME_MS,
|
||||
TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
/**
|
||||
* 通知机器人开启对讲。
|
||||
*/
|
||||
public void sendStart(String operatorId) {
|
||||
RobotAudio.RobotAudioFormat format = RobotAudio.RobotAudioFormat.newBuilder()
|
||||
.setSampleRate(SAMPLE_RATE)
|
||||
.setChannels(CHANNELS)
|
||||
.setBitsPerSample(BITS_PER_SAMPLE)
|
||||
.setEncoding(ENCODING)
|
||||
.build();
|
||||
|
||||
RobotAudio.RobotAudioStart start = RobotAudio.RobotAudioStart.newBuilder()
|
||||
.setSessionId(sessionId)
|
||||
.setOperatorId(operatorId == null ? "" : operatorId)
|
||||
.setFormat(format)
|
||||
.build();
|
||||
|
||||
log.info("gRPC下发终端音频协商格式，sessionId={}，terminalId={}，deviceId={}，sampleRate={}，channels={}，bitsPerSample={}，encoding={}，frameMs={}，frameSamples={}，frameBytes={}",
|
||||
sessionId,
|
||||
terminalId,
|
||||
deviceId,
|
||||
SAMPLE_RATE,
|
||||
CHANNELS,
|
||||
BITS_PER_SAMPLE,
|
||||
ENCODING,
|
||||
FRAME_MS,
|
||||
FRAME_SAMPLES,
|
||||
FRAME_BYTES);
|
||||
|
||||
send(RobotAudio.RobotAudioRequest.newBuilder()
|
||||
.setHeader(EdgeCommonUtil.buildRequest(deviceId))
|
||||
.setStart(start)
|
||||
.build());
|
||||
}
|
||||
|
||||
/**
|
||||
* 将浏览器侧 WebRTC 媒体通道产出的 PCM 帧发送给机器人。
|
||||
*/
|
||||
public void sendAudio(byte[] pcm) {
|
||||
if (pcm == null || pcm.length == 0) {
|
||||
return;
|
||||
}
|
||||
if (closed.get()) {
|
||||
return;
|
||||
}
|
||||
enqueueBrowserAudio(pcm);
|
||||
}
|
||||
|
||||
private void enqueueBrowserAudio(byte[] pcm) {
|
||||
if (closed.get()) {
|
||||
return;
|
||||
}
|
||||
synchronized (browserAudioBufferLock) {
|
||||
int offset = 0;
|
||||
if (browserAudioRemainderLength > 0) {
|
||||
int need = FRAME_BYTES - browserAudioRemainderLength;
|
||||
int copy = Math.min(need, pcm.length);
|
||||
System.arraycopy(pcm, 0, browserAudioRemainder, browserAudioRemainderLength, copy);
|
||||
browserAudioRemainderLength += copy;
|
||||
offset += copy;
|
||||
if (browserAudioRemainderLength == FRAME_BYTES) {
|
||||
offerBrowserAudioFrame(browserAudioRemainder);
|
||||
browserAudioRemainder = new byte[FRAME_BYTES];
|
||||
browserAudioRemainderLength = 0;
|
||||
}
|
||||
}
|
||||
|
||||
while (offset + FRAME_BYTES <= pcm.length) {
|
||||
byte[] frame = new byte[FRAME_BYTES];
|
||||
System.arraycopy(pcm, offset, frame, 0, FRAME_BYTES);
|
||||
offerBrowserAudioFrame(frame);
|
||||
offset += FRAME_BYTES;
|
||||
}
|
||||
|
||||
int remain = pcm.length - offset;
|
||||
if (remain > 0) {
|
||||
System.arraycopy(pcm, offset, browserAudioRemainder, 0, remain);
|
||||
browserAudioRemainderLength = remain;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void offerBrowserAudioFrame(byte[] frame) {
|
||||
if (queuedBrowserFrames.get() >= MAX_QUEUED_BROWSER_FRAMES) {
|
||||
if (browserAudioQueue.poll() != null) {
|
||||
queuedBrowserFrames.decrementAndGet();
|
||||
browserAudioFramesDropped++;
|
||||
}
|
||||
}
|
||||
browserAudioQueue.offer(frame.clone());
|
||||
queuedBrowserFrames.incrementAndGet();
|
||||
browserAudioFramesEnqueued++;
|
||||
browserAudioStarted = true;
|
||||
logBrowserAudioStatsIfNeeded("browserAudioEnqueue");
|
||||
}
|
||||
|
||||
private void sendQueuedBrowserAudio() {
|
||||
if (closed.get()) {
|
||||
return;
|
||||
}
|
||||
trimBrowserAudioQueue();
|
||||
byte[] pcm = browserAudioQueue.poll();
|
||||
if (pcm == null) {
|
||||
if (!browserAudioStarted) {
|
||||
return;
|
||||
}
|
||||
pcm = new byte[FRAME_BYTES];
|
||||
browserAudioSilentFramesSent++;
|
||||
} else {
|
||||
queuedBrowserFrames.decrementAndGet();
|
||||
}
|
||||
|
||||
RobotAudio.RobotAudioFrame frame = RobotAudio.RobotAudioFrame.newBuilder()
|
||||
.setSessionId(sessionId)
|
||||
.setSequence(sequence.incrementAndGet())
|
||||
.setTimestampMs(System.currentTimeMillis())
|
||||
.setPcm(ByteString.copyFrom(pcm))
|
||||
.build();
|
||||
logGrpcOutboundFormatIfNeeded(frame);
|
||||
|
||||
sendDirect(RobotAudio.RobotAudioRequest.newBuilder()
|
||||
.setHeader(EdgeCommonUtil.buildRequest(deviceId))
|
||||
.setAudioFrame(frame)
|
||||
.build());
|
||||
browserAudioFramesSent++;
|
||||
logBrowserAudioStatsIfNeeded("browserAudioGrpcSend");
|
||||
}
|
||||
|
||||
private void trimBrowserAudioQueue() {
|
||||
while (queuedBrowserFrames.get() > TARGET_QUEUED_BROWSER_FRAMES) {
|
||||
byte[] dropped = browserAudioQueue.poll();
|
||||
if (dropped == null) {
|
||||
queuedBrowserFrames.set(0);
|
||||
break;
|
||||
}
|
||||
queuedBrowserFrames.decrementAndGet();
|
||||
browserAudioFramesDropped++;
|
||||
}
|
||||
}
|
||||
|
||||
private void logBrowserAudioStatsIfNeeded(String source) {
|
||||
long now = System.currentTimeMillis();
|
||||
if (now - lastAudioStatsLogAt < AUDIO_STATS_LOG_INTERVAL_MS) {
|
||||
return;
|
||||
}
|
||||
lastAudioStatsLogAt = now;
|
||||
log.info("gRPC浏览器音频统计，source={}，sessionId={}，enqueued={}，sent={}，dropped={}，queued={}",
|
||||
source,
|
||||
sessionId,
|
||||
browserAudioFramesEnqueued,
|
||||
browserAudioFramesSent,
|
||||
browserAudioFramesDropped,
|
||||
queuedBrowserFrames.get());
|
||||
log.info("GRPC-BROWSER-AUDIO-STATS source={} sessionId={} enqueued={} sent={} dropped={} silentSent={} queued={}",
|
||||
source,
|
||||
sessionId,
|
||||
browserAudioFramesEnqueued,
|
||||
browserAudioFramesSent,
|
||||
browserAudioFramesDropped,
|
||||
browserAudioSilentFramesSent,
|
||||
queuedBrowserFrames.get());
|
||||
}
|
||||
|
||||
private void logGrpcOutboundFormatIfNeeded(RobotAudio.RobotAudioFrame frame) {
|
||||
long now = System.currentTimeMillis();
|
||||
if (now - lastGrpcOutboundFormatLogAt < AUDIO_STATS_LOG_INTERVAL_MS) {
|
||||
return;
|
||||
}
|
||||
lastGrpcOutboundFormatLogAt = now;
|
||||
int bytes = frame.getPcm().size();
|
||||
int frameSamples = bytes / (CHANNELS * (BITS_PER_SAMPLE / 8));
|
||||
int frameMs = frameSamples * 1000 / SAMPLE_RATE;
|
||||
log.info("gRPC浏览器下行音频格式，sessionId={}，sampleRate={}，channels={}，bitsPerSample={}，encoding={}，frameSamples={}，frameMs={}，bytes={}，sequence={}，timestampMs={}",
|
||||
sessionId,
|
||||
SAMPLE_RATE,
|
||||
CHANNELS,
|
||||
BITS_PER_SAMPLE,
|
||||
ENCODING,
|
||||
frameSamples,
|
||||
frameMs,
|
||||
bytes,
|
||||
frame.getSequence(),
|
||||
frame.getTimestampMs());
|
||||
logPcmStats("grpc-outbound-browser", sessionId, frame.getPcm().toByteArray(), frame.getSequence(), frame.getTimestampMs());
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Java 侧保活心跳。机器人仍需按协议每秒上报心跳。
|
||||
*/
|
||||
public void sendHeartbeat() {
|
||||
RobotAudio.RobotAudioHeartbeat heartbeat = RobotAudio.RobotAudioHeartbeat.newBuilder()
|
||||
.setSessionId(sessionId)
|
||||
.setTimestampMs(System.currentTimeMillis())
|
||||
.build();
|
||||
|
||||
send(RobotAudio.RobotAudioRequest.newBuilder()
|
||||
.setHeader(EdgeCommonUtil.buildRequest(deviceId))
|
||||
.setHeartbeat(heartbeat)
|
||||
.build());
|
||||
}
|
||||
|
||||
/**
|
||||
* 正常结束对讲。
|
||||
*/
|
||||
public void sendStop(String reason) {
|
||||
if (!closed.compareAndSet(false, true)) {
|
||||
return;
|
||||
}
|
||||
RobotAudio.RobotAudioStop stop = RobotAudio.RobotAudioStop.newBuilder()
|
||||
.setSessionId(sessionId)
|
||||
.setReason(reason == null ? "" : reason)
|
||||
.build();
|
||||
RobotAudio.RobotAudioRequest request = RobotAudio.RobotAudioRequest.newBuilder()
|
||||
.setHeader(EdgeCommonUtil.buildRequest(deviceId))
|
||||
.setStop(stop)
|
||||
.build();
|
||||
executeOnGrpcSendThread(() -> {
|
||||
sendDirect(request);
|
||||
completeDirect();
|
||||
shutdownBrowserAudioSender();
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 异常关闭流。
|
||||
*/
|
||||
public void cancel(String reason) {
|
||||
if (closed.compareAndSet(false, true)) {
|
||||
executeOnGrpcSendThread(() -> {
|
||||
synchronized (sendLock) {
|
||||
requestObserver.onError(Status.CANCELLED.withDescription(reason).asRuntimeException());
|
||||
}
|
||||
shutdownBrowserAudioSender();
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private void complete() {
|
||||
if (closed.compareAndSet(false, true)) {
|
||||
executeOnGrpcSendThread(() -> {
|
||||
completeDirect();
|
||||
shutdownBrowserAudioSender();
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private void completeDirect() {
|
||||
synchronized (sendLock) {
|
||||
requestObserver.onCompleted();
|
||||
}
|
||||
}
|
||||
|
||||
private void shutdownBrowserAudioSender() {
|
||||
if (browserAudioSenderFuture != null) {
|
||||
browserAudioSenderFuture.cancel(false);
|
||||
}
|
||||
browserAudioQueue.clear();
|
||||
queuedBrowserFrames.set(0);
|
||||
browserAudioSender.shutdown();
|
||||
}
|
||||
|
||||
private void send(RobotAudio.RobotAudioRequest request) {
|
||||
if (closed.get()) {
|
||||
log.warn("机器人音频流已关闭，忽略发送，sessionId={}", sessionId);
|
||||
return;
|
||||
}
|
||||
executeOnGrpcSendThread(() -> {
|
||||
if (closed.get()) {
|
||||
return;
|
||||
}
|
||||
sendDirect(request);
|
||||
});
|
||||
}
|
||||
|
||||
private void executeOnGrpcSendThread(Runnable runnable) {
|
||||
try {
|
||||
browserAudioSender.execute(runnable);
|
||||
} catch (Exception e) {
|
||||
log.warn("机器人音频gRPC发送线程不可用，sessionId={}", sessionId, e);
|
||||
}
|
||||
}
|
||||
|
||||
private void sendDirect(RobotAudio.RobotAudioRequest request) {
|
||||
synchronized (sendLock) {
|
||||
requestObserver.onNext(request);
|
||||
}
|
||||
}
|
||||
|
||||
public String getTerminalId() {
|
||||
return terminalId;
|
||||
}
|
||||
|
||||
public String getDeviceId() {
|
||||
return deviceId;
|
||||
}
|
||||
|
||||
public String getSessionId() {
|
||||
return sessionId;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* gRPC 流事件监听器，由业务层实现会话状态更新和音频转发。
|
||||
*/
|
||||
public interface RobotAudioStreamListener {
|
||||
void onStarted(String sessionId, boolean accepted, String message);
|
||||
|
||||
void onRobotAudio(String sessionId, long sequence, long timestampMs, byte[] pcm);
|
||||
|
||||
void onHeartbeat(String sessionId, long timestampMs);
|
||||
|
||||
void onRobotStop(String sessionId, String reason);
|
||||
|
||||
void onError(String sessionId, Throwable throwable);
|
||||
|
||||
void onCompleted(String sessionId);
|
||||
}
|
||||
|
||||
private static class WavDumpRecorder {
|
||||
private final String name;
|
||||
private final int maxBytes;
|
||||
private final ByteArrayOutputStream buffer;
|
||||
private Path filePath;
|
||||
private int writtenBytes;
|
||||
private boolean completed;
|
||||
|
||||
private WavDumpRecorder(String name, int maxBytes) {
|
||||
this.name = name;
|
||||
this.maxBytes = maxBytes;
|
||||
this.buffer = new ByteArrayOutputStream(maxBytes);
|
||||
}
|
||||
|
||||
private synchronized void record(byte[] pcm) {
|
||||
if (completed || pcm == null || pcm.length == 0) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
if (filePath == null) {
|
||||
Path dir = Paths.get("logs", "audio-dump");
|
||||
Files.createDirectories(dir);
|
||||
filePath = dir.resolve(name + "-" + System.currentTimeMillis() + ".wav");
|
||||
log.info("开始保存10秒WAV音频，name={}，file={}，sampleRate={}，channels={}，bitsPerSample={}",
|
||||
name, filePath.toAbsolutePath(), SAMPLE_RATE, CHANNELS, BITS_PER_SAMPLE);
|
||||
}
|
||||
int remain = maxBytes - writtenBytes;
|
||||
int writeBytes = Math.min(remain, pcm.length);
|
||||
buffer.write(pcm, 0, writeBytes);
|
||||
writtenBytes += writeBytes;
|
||||
if (writtenBytes >= maxBytes) {
|
||||
writeWavFile(filePath, buffer.toByteArray(), SAMPLE_RATE, CHANNELS, BITS_PER_SAMPLE);
|
||||
completed = true;
|
||||
log.info("WAV音频保存完成，name={}，file={}，pcmBytes={}",
|
||||
name, filePath == null ? "" : filePath.toAbsolutePath(), writtenBytes);
|
||||
}
|
||||
} catch (IOException e) {
|
||||
completed = true;
|
||||
log.warn("保存WAV音频失败，name={}，file={}", name, filePath, e);
|
||||
}
|
||||
}
|
||||
|
||||
private void writeWavFile(Path path, byte[] pcm, int sampleRate, int channels, int bitsPerSample) throws IOException {
|
||||
try (OutputStream out = Files.newOutputStream(path, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE)) {
|
||||
writeAscii(out, "RIFF");
|
||||
writeIntLe(out, 36 + pcm.length);
|
||||
writeAscii(out, "WAVE");
|
||||
writeAscii(out, "fmt ");
|
||||
writeIntLe(out, 16);
|
||||
writeShortLe(out, 1);
|
||||
writeShortLe(out, channels);
|
||||
writeIntLe(out, sampleRate);
|
||||
writeIntLe(out, sampleRate * channels * bitsPerSample / 8);
|
||||
writeShortLe(out, channels * bitsPerSample / 8);
|
||||
writeShortLe(out, bitsPerSample);
|
||||
writeAscii(out, "data");
|
||||
writeIntLe(out, pcm.length);
|
||||
out.write(pcm);
|
||||
}
|
||||
}
|
||||
|
||||
private void writeAscii(OutputStream out, String value) throws IOException {
|
||||
out.write(value.getBytes(java.nio.charset.StandardCharsets.US_ASCII));
|
||||
}
|
||||
|
||||
private void writeIntLe(OutputStream out, int value) throws IOException {
|
||||
out.write(value & 0xFF);
|
||||
out.write((value >> 8) & 0xFF);
|
||||
out.write((value >> 16) & 0xFF);
|
||||
out.write((value >> 24) & 0xFF);
|
||||
}
|
||||
|
||||
private void writeShortLe(OutputStream out, int value) throws IOException {
|
||||
out.write(value & 0xFF);
|
||||
out.write((value >> 8) & 0xFF);
|
||||
}
|
||||
}
|
||||
|
||||
private static class PcmDumpRecorder {
|
||||
private final String name;
|
||||
private final int maxBytes;
|
||||
private OutputStream outputStream;
|
||||
private Path filePath;
|
||||
private int writtenBytes;
|
||||
private boolean completed;
|
||||
|
||||
private PcmDumpRecorder(String name, int maxBytes) {
|
||||
this.name = name;
|
||||
this.maxBytes = maxBytes;
|
||||
}
|
||||
|
||||
private synchronized void record(byte[] pcm) {
|
||||
if (completed || pcm == null || pcm.length == 0) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
if (outputStream == null) {
|
||||
Path dir = Paths.get("logs", "audio-dump");
|
||||
Files.createDirectories(dir);
|
||||
filePath = dir.resolve(name + "-" + System.currentTimeMillis() + ".pcm");
|
||||
outputStream = Files.newOutputStream(filePath, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE);
|
||||
log.info("开始保存10秒PCM音频，name={}，file={}，sampleRate={}，channels={}，bitsPerSample={}",
|
||||
name, filePath.toAbsolutePath(), SAMPLE_RATE, CHANNELS, BITS_PER_SAMPLE);
|
||||
}
|
||||
int remain = maxBytes - writtenBytes;
|
||||
int writeBytes = Math.min(remain, pcm.length);
|
||||
outputStream.write(pcm, 0, writeBytes);
|
||||
writtenBytes += writeBytes;
|
||||
if (writtenBytes >= maxBytes) {
|
||||
close();
|
||||
completed = true;
|
||||
log.info("PCM音频保存完成，name={}，file={}，bytes={}",
|
||||
name, filePath == null ? "" : filePath.toAbsolutePath(), writtenBytes);
|
||||
}
|
||||
} catch (IOException e) {
|
||||
completed = true;
|
||||
closeQuietly();
|
||||
log.warn("保存PCM音频失败，name={}，file={}", name, filePath, e);
|
||||
}
|
||||
}
|
||||
|
||||
private void close() throws IOException {
|
||||
if (outputStream != null) {
|
||||
outputStream.flush();
|
||||
outputStream.close();
|
||||
outputStream = null;
|
||||
}
|
||||
}
|
||||
|
||||
private void closeQuietly() {
|
||||
try {
|
||||
close();
|
||||
} catch (IOException ignored) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static class PcmStats {
|
||||
private final int samples;
|
||||
private final int rms;
|
||||
private final int peak;
|
||||
private final String zeroCrossRate;
|
||||
private final String firstSamples;
|
||||
|
||||
private PcmStats(int samples, int rms, int peak, String zeroCrossRate, String firstSamples) {
|
||||
this.samples = samples;
|
||||
this.rms = rms;
|
||||
this.peak = peak;
|
||||
this.zeroCrossRate = zeroCrossRate;
|
||||
this.firstSamples = firstSamples;
|
||||
}
|
||||
|
||||
private static PcmStats fromPcm16Le(byte[] pcm) {
|
||||
if (pcm == null || pcm.length < 2) {
|
||||
return new PcmStats(0, 0, 0, "0.0000", "[]");
|
||||
}
|
||||
|
||||
int samples = pcm.length / 2;
|
||||
long squareSum = 0L;
|
||||
int peak = 0;
|
||||
int zeroCross = 0;
|
||||
int previous = 0;
|
||||
StringBuilder first = new StringBuilder("[");
|
||||
int firstCount = Math.min(8, samples);
|
||||
|
||||
for (int i = 0; i < samples; i++) {
|
||||
int low = pcm[i * 2] & 0xFF;
|
||||
int high = pcm[i * 2 + 1];
|
||||
short sample = (short) ((high << 8) | low);
|
||||
int value = sample;
|
||||
int abs = Math.abs(value);
|
||||
peak = Math.max(peak, abs);
|
||||
squareSum += (long) value * value;
|
||||
if (i > 0 && ((previous < 0 && value >= 0) || (previous >= 0 && value < 0))) {
|
||||
zeroCross++;
|
||||
}
|
||||
previous = value;
|
||||
if (i < firstCount) {
|
||||
if (i > 0) {
|
||||
first.append(',');
|
||||
}
|
||||
first.append(value);
|
||||
}
|
||||
}
|
||||
first.append(']');
|
||||
|
||||
int rms = (int) Math.round(Math.sqrt(squareSum / (double) samples));
|
||||
String zeroCrossRate = String.format(java.util.Locale.ROOT, "%.4f", zeroCross / (double) Math.max(1, samples - 1));
|
||||
return new PcmStats(samples, rms, peak, zeroCrossRate, first.toString());
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -59,6 +59,8 @@ public class GrpcServiceManager {
|
||||
clientFactories.put(HlcServiceGrpc.HlcServiceBlockingStub.class, new GrpcClientFactory<>(HlcServiceGrpc::newBlockingStub));
|
||||
// 注册agv服务的stub
|
||||
clientFactories.put(AgvServiceGrpc.AgvServiceBlockingStub.class, new GrpcClientFactory<>(AgvServiceGrpc::newBlockingStub));
|
||||
// 注册机器人实时语音双向流服务的异步Stub
|
||||
clientFactories.put(RobotAudioServiceGrpc.RobotAudioServiceStub.class, new GrpcClientFactory<>(RobotAudioServiceGrpc::newStub));
|
||||
}
|
||||
|
||||
|
||||
|
||||
@ -0,0 +1,591 @@
|
||||
package com.cmvr.edge.client.service;
|
||||
|
||||
import com.cmvr.common.utils.uuid.IdUtils;
|
||||
import com.cmvr.edge.client.adapter.RobotAudioGrpcAdapter;
|
||||
import com.cmvr.edge.client.adapter.RobotAudioGrpcAdapter.RobotAudioStream;
|
||||
import com.cmvr.edge.client.adapter.RobotAudioGrpcAdapter.RobotAudioStreamListener;
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.databind.node.ObjectNode;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.web.socket.BinaryMessage;
|
||||
import org.springframework.web.socket.TextMessage;
|
||||
import org.springframework.web.socket.WebSocketSession;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.Iterator;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
/**
|
||||
* 实时语音对讲业务层。
|
||||
*
|
||||
* 分工：
|
||||
* 1. WebSocket 只处理 WebRTC 信令 JSON，不承载音频；
|
||||
* 2. 浏览器 WebRTC 媒体通道产出的 PCM 通过 sendBrowserAudio 方法进入业务层；
|
||||
* 3. 业务层通过 RobotAudioGrpcAdapter 将 PCM 转发到机器人 gRPC 双向流；
|
||||
* 4. 机器人音频通过 gRPC 回调给业务层，再交给已接入的 WebRTC 媒体发送侧。
|
||||
*/
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class AudioService implements RobotAudioStreamListener {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(AudioService.class);
|
||||
|
||||
/** 机器人心跳超时时间：机器人每 1 秒上报一次，超过 5 秒判定离线。 */
|
||||
private static final long ROBOT_HEARTBEAT_TIMEOUT_MS = 5000L;
|
||||
|
||||
private final RobotAudioGrpcAdapter robotAudioGrpcAdapter;
|
||||
|
||||
private final ObjectMapper objectMapper = new ObjectMapper();
|
||||
|
||||
/** sessionId -> 会话上下文。 */
|
||||
private final ConcurrentHashMap<String, AudioSession> sessionMap = new ConcurrentHashMap<>();
|
||||
private final ConcurrentHashMap<String, String> webSocketSessionMap = new ConcurrentHashMap<>();
|
||||
|
||||
/**
|
||||
* 处理 /ws/signaling 收到的 JSON 信令。
|
||||
*/
|
||||
public void handleSignalingMessage(WebSocketSession webSocketSession, String payload) {
|
||||
try {
|
||||
JsonNode jsonNode = objectMapper.readTree(payload);
|
||||
String type = getText(jsonNode, "type");
|
||||
if (type == null) {
|
||||
sendError(webSocketSession, null, "信令缺少type字段");
|
||||
return;
|
||||
}
|
||||
|
||||
switch (type) {
|
||||
case "join":
|
||||
case "start":
|
||||
handleJoin(webSocketSession, jsonNode);
|
||||
break;
|
||||
case "offer":
|
||||
case "answer":
|
||||
case "candidate":
|
||||
sendError(webSocketSession, getText(jsonNode, "sessionId"), "WebRTC disabled, use WebSocket binary audio");
|
||||
break;
|
||||
case "leave":
|
||||
case "stop":
|
||||
stopSession(getText(jsonNode, "sessionId"), "browser stop");
|
||||
break;
|
||||
case "heartbeat":
|
||||
handleBrowserHeartbeat(webSocketSession, jsonNode);
|
||||
break;
|
||||
default:
|
||||
sendError(webSocketSession, getText(jsonNode, "sessionId"), "未知信令类型：" + type);
|
||||
break;
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("处理WebRTC信令失败，payload={}", payload, e);
|
||||
sendError(webSocketSession, null, "处理信令失败：" + e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 浏览器 WebSocket 断开时关闭对应会话。
|
||||
*/
|
||||
public void handleWebSocketClosed(WebSocketSession webSocketSession) {
|
||||
if (webSocketSession == null) {
|
||||
return;
|
||||
}
|
||||
String wsId = webSocketSession.getId();
|
||||
webSocketSessionMap.remove(wsId);
|
||||
for (AudioSession session : sessionMap.values()) {
|
||||
if (session.webSocketSession != null && wsId.equals(session.webSocketSession.getId())) {
|
||||
stopSession(session.sessionId, "websocket closed");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 浏览器通过 WebSocket BinaryMessage 发送 PCM_S16LE / 48k / mono 音频。
|
||||
* 推荐前端按 10ms 一帧发送：480 samples，960 bytes。
|
||||
*/
|
||||
public void handleBrowserAudioMessage(WebSocketSession webSocketSession, ByteBuffer payload) {
|
||||
if (webSocketSession == null || payload == null) {
|
||||
return;
|
||||
}
|
||||
String sessionId = webSocketSessionMap.get(webSocketSession.getId());
|
||||
if (sessionId == null) {
|
||||
sessionId = findSessionIdByWebSocket(webSocketSession);
|
||||
}
|
||||
if (sessionId == null) {
|
||||
log.warn("收到浏览器音频但未找到会话，wsSessionId={}，bytes={}", webSocketSession.getId(), payload.remaining());
|
||||
return;
|
||||
}
|
||||
byte[] pcm = new byte[payload.remaining()];
|
||||
payload.get(pcm);
|
||||
sendBrowserAudio(sessionId, pcm);
|
||||
}
|
||||
|
||||
/**
|
||||
* WebRTC 媒体层调用此方法，把浏览器侧 PCM-48K-16bit-单声道音频帧发给机器人。
|
||||
*
|
||||
* 注意：WebSocket 禁止传音频，音频必须来自 WebRTC 媒体通道。
|
||||
*/
|
||||
public void sendBrowserAudio(String sessionId, byte[] pcm) {
|
||||
AudioSession session = sessionMap.get(sessionId);
|
||||
if (session == null) {
|
||||
log.warn("浏览器音频帧找不到会话，sessionId={}", sessionId);
|
||||
return;
|
||||
}
|
||||
try {
|
||||
session.lastBrowserAudioAt = System.currentTimeMillis();
|
||||
session.robotStream.sendAudio(pcm);
|
||||
} catch (Exception e) {
|
||||
log.error("发送浏览器音频到机器人失败，sessionId={}", sessionId, e);
|
||||
failSession(session, "发送浏览器音频失败：" + e.getMessage(), true);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* WebRTC 媒体发送层可注册该回调，接收机器人 gRPC 上传的 PCM 音频。
|
||||
*/
|
||||
public void registerRobotAudioConsumer(String sessionId, Consumer<byte[]> robotAudioConsumer) {
|
||||
AudioSession session = sessionMap.get(sessionId);
|
||||
if (session == null) {
|
||||
throw new IllegalArgumentException("会话不存在：" + sessionId);
|
||||
}
|
||||
session.robotAudioConsumer = robotAudioConsumer;
|
||||
}
|
||||
|
||||
public void unregisterRobotAudioConsumer(String sessionId) {
|
||||
AudioSession session = sessionMap.get(sessionId);
|
||||
if (session != null) {
|
||||
session.robotAudioConsumer = null;
|
||||
}
|
||||
}
|
||||
|
||||
private void handleJoin(WebSocketSession webSocketSession, JsonNode jsonNode) {
|
||||
String terminalId = getText(jsonNode, "terminalId");
|
||||
String deviceId = getText(jsonNode, "deviceId");
|
||||
String operatorId = getText(jsonNode, "operatorId");
|
||||
String sessionId = getText(jsonNode, "sessionId");
|
||||
|
||||
if (terminalId == null || deviceId == null) {
|
||||
sendError(webSocketSession, sessionId, "terminalId和deviceId不能为空");
|
||||
return;
|
||||
}
|
||||
if (sessionId == null) {
|
||||
sessionId = IdUtils.fastSimpleUUID();
|
||||
}
|
||||
|
||||
try {
|
||||
AudioSession exists = sessionMap.get(sessionId);
|
||||
if (exists != null) {
|
||||
exists.webSocketSession = webSocketSession;
|
||||
exists.lastBrowserSignalAt = System.currentTimeMillis();
|
||||
webSocketSessionMap.put(webSocketSession.getId(), sessionId);
|
||||
sendState(exists, "joined");
|
||||
return;
|
||||
}
|
||||
|
||||
AudioSession session = new AudioSession();
|
||||
session.sessionId = sessionId;
|
||||
session.terminalId = terminalId;
|
||||
session.deviceId = deviceId;
|
||||
session.operatorId = operatorId;
|
||||
session.webSocketSession = webSocketSession;
|
||||
session.state = AudioSessionState.CONNECTING;
|
||||
session.createdAt = System.currentTimeMillis();
|
||||
session.lastBrowserSignalAt = session.createdAt;
|
||||
session.lastRobotHeartbeatAt = session.createdAt;
|
||||
|
||||
sessionMap.put(sessionId, session);
|
||||
webSocketSessionMap.put(webSocketSession.getId(), sessionId);
|
||||
try {
|
||||
session.robotStream = robotAudioGrpcAdapter.openStream(terminalId, deviceId, sessionId, operatorId, this);
|
||||
} catch (Exception e) {
|
||||
sessionMap.remove(sessionId);
|
||||
webSocketSessionMap.remove(webSocketSession.getId());
|
||||
throw e;
|
||||
}
|
||||
sendState(session, "joined");
|
||||
log.info("实时语音会话创建成功，sessionId={}，terminalId={}，deviceId={}", sessionId, terminalId, deviceId);
|
||||
} catch (Exception e) {
|
||||
log.error("创建实时语音会话失败，sessionId={}", sessionId, e);
|
||||
sendError(webSocketSession, sessionId, "创建实时语音会话失败：" + e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private void handleWebRtcSignal(WebSocketSession webSocketSession, JsonNode jsonNode, String type) {
|
||||
String sessionId = getText(jsonNode, "sessionId");
|
||||
|
||||
// 如果sessionId为空，尝试从WebSocket会话关联的session中查找
|
||||
if (sessionId == null) {
|
||||
for (AudioSession session : sessionMap.values()) {
|
||||
if (session.webSocketSession != null &&
|
||||
session.webSocketSession.getId().equals(webSocketSession.getId())) {
|
||||
sessionId = session.sessionId;
|
||||
log.warn("WebRTC信令缺少sessionId，自动关联到会话: {}", sessionId);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
AudioSession session = sessionMap.get(sessionId);
|
||||
if (session == null) {
|
||||
sendError(webSocketSession, sessionId, "会话不存在，请先发送join信令");
|
||||
return;
|
||||
}
|
||||
|
||||
session.webSocketSession = webSocketSession;
|
||||
session.lastBrowserSignalAt = System.currentTimeMillis();
|
||||
|
||||
try {
|
||||
if ("offer".equals(type)) {
|
||||
String sdp = getSdp(jsonNode);
|
||||
if (sdp == null) {
|
||||
sendError(webSocketSession, sessionId, "offer信令缺少sdp字段");
|
||||
return;
|
||||
}
|
||||
sendError(webSocketSession, sessionId, "WebRTC disabled, use WebSocket binary audio");
|
||||
} else if ("candidate".equals(type)) {
|
||||
IceCandidateValue candidate = getIceCandidate(jsonNode);
|
||||
if (candidate == null || candidate.candidate == null) {
|
||||
sendError(webSocketSession, sessionId, "candidate信令格式错误");
|
||||
return;
|
||||
}
|
||||
sendError(webSocketSession, sessionId, "WebRTC disabled, use WebSocket binary audio");
|
||||
} else {
|
||||
ObjectNode ack = baseMessage("signalAck", sessionId);
|
||||
ack.put("signalType", type);
|
||||
ack.put("state", session.state.name());
|
||||
sendJson(webSocketSession, ack);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("处理WebRTC媒体信令失败，sessionId={}，type={}", sessionId, type, e);
|
||||
sendError(webSocketSession, sessionId, "处理WebRTC媒体信令失败：" + e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private void handleBrowserHeartbeat(WebSocketSession webSocketSession, JsonNode jsonNode) {
|
||||
String sessionId = getText(jsonNode, "sessionId");
|
||||
AudioSession session = sessionMap.get(sessionId);
|
||||
if (session == null) {
|
||||
sendError(webSocketSession, sessionId, "会话不存在");
|
||||
return;
|
||||
}
|
||||
session.lastBrowserSignalAt = System.currentTimeMillis();
|
||||
if (session.robotStream != null) {
|
||||
session.robotStream.sendHeartbeat();
|
||||
}
|
||||
|
||||
ObjectNode pong = baseMessage("heartbeatAck", sessionId);
|
||||
pong.put("state", session.state.name());
|
||||
sendJson(webSocketSession, pong);
|
||||
}
|
||||
|
||||
public void stopSession(String sessionId, String reason) {
|
||||
if (sessionId == null) {
|
||||
return;
|
||||
}
|
||||
AudioSession session = sessionMap.remove(sessionId);
|
||||
if (session == null) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
session.state = AudioSessionState.CLOSED;
|
||||
removeWebSocketMapping(session);
|
||||
if (session.robotStream != null) {
|
||||
session.robotStream.sendStop(reason);
|
||||
}
|
||||
sendState(session, "closed");
|
||||
log.info("实时语音会话已关闭，sessionId={}，reason={}", sessionId, reason);
|
||||
} catch (Exception e) {
|
||||
log.error("关闭实时语音会话异常，sessionId={}", sessionId, e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 定时检测机器人心跳。
|
||||
*/
|
||||
@Scheduled(fixedDelay = 1000L)
|
||||
public void checkRobotHeartbeatTimeout() {
|
||||
long now = System.currentTimeMillis();
|
||||
Iterator<Map.Entry<String, AudioSession>> iterator = sessionMap.entrySet().iterator();
|
||||
while (iterator.hasNext()) {
|
||||
Map.Entry<String, AudioSession> entry = iterator.next();
|
||||
AudioSession session = entry.getValue();
|
||||
if (session.state == AudioSessionState.CLOSED || session.state == AudioSessionState.FAILED) {
|
||||
continue;
|
||||
}
|
||||
if (now - session.lastRobotHeartbeatAt > ROBOT_HEARTBEAT_TIMEOUT_MS) {
|
||||
log.warn("机器人心跳超时，sessionId={}，terminalId={}，deviceId={}",
|
||||
session.sessionId, session.terminalId, session.deviceId);
|
||||
// 先从业务会话表摘除，避免 cancel/close 触发的异步回调再次进入同一个会话并重复关闭 native 对象。
|
||||
iterator.remove();
|
||||
session.state = AudioSessionState.ROBOT_OFFLINE;
|
||||
sendState(session, "robotOffline");
|
||||
if (session.robotStream != null) {
|
||||
session.robotStream.cancel("robot heartbeat timeout");
|
||||
}
|
||||
removeWebSocketMapping(session);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onStarted(String sessionId, boolean accepted, String message) {
|
||||
AudioSession session = sessionMap.get(sessionId);
|
||||
if (session == null) {
|
||||
return;
|
||||
}
|
||||
session.lastRobotHeartbeatAt = System.currentTimeMillis();
|
||||
session.state = accepted ? AudioSessionState.RUNNING : AudioSessionState.FAILED;
|
||||
ObjectNode event = baseMessage("robotStartAck", sessionId);
|
||||
event.put("accepted", accepted);
|
||||
event.put("message", message == null ? "" : message);
|
||||
event.put("state", session.state.name());
|
||||
sendJson(session.webSocketSession, event);
|
||||
if (!accepted) {
|
||||
failSession(session, message == null ? "机器人拒绝开启语音对讲" : message, true);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onRobotAudio(String sessionId, long sequence, long timestampMs, byte[] pcm) {
|
||||
AudioSession session = sessionMap.get(sessionId);
|
||||
if (session == null) {
|
||||
// log.warn("机器人音频帧找不到会话，sessionId={}", sessionId);
|
||||
return;
|
||||
}
|
||||
session.lastRobotAudioAt = System.currentTimeMillis();
|
||||
session.lastRobotHeartbeatAt = session.lastRobotAudioAt;
|
||||
sendBinary(session.webSocketSession, pcm);
|
||||
Consumer<byte[]> consumer = session.robotAudioConsumer;
|
||||
if (consumer != null) {
|
||||
try {
|
||||
consumer.accept(pcm);
|
||||
} catch (Exception e) {
|
||||
log.error("投递机器人音频到WebRTC媒体层失败，sessionId={}，sequence={}", sessionId, sequence, e);
|
||||
}
|
||||
} else {
|
||||
log.debug("未注册机器人音频消费者，丢弃PCM帧，sessionId={}，sequence={}，bytes={}",
|
||||
sessionId, sequence, pcm == null ? 0 : pcm.length);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onHeartbeat(String sessionId, long timestampMs) {
|
||||
AudioSession session = sessionMap.get(sessionId);
|
||||
if (session == null) {
|
||||
return;
|
||||
}
|
||||
session.lastRobotHeartbeatAt = System.currentTimeMillis();
|
||||
if (session.state == AudioSessionState.CONNECTING) {
|
||||
session.state = AudioSessionState.RUNNING;
|
||||
sendState(session, "running");
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onRobotStop(String sessionId, String reason) {
|
||||
AudioSession session = sessionMap.get(sessionId);
|
||||
if (session != null) {
|
||||
sendState(session, "robotStop");
|
||||
}
|
||||
stopSession(sessionId, reason == null ? "robot stop" : reason);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(String sessionId, Throwable throwable) {
|
||||
AudioSession session = sessionMap.get(sessionId);
|
||||
if (session == null) {
|
||||
return;
|
||||
}
|
||||
failSession(session, throwable == null ? "机器人音频流异常" : throwable.getMessage(), false);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted(String sessionId) {
|
||||
AudioSession session = sessionMap.get(sessionId);
|
||||
if (session != null) {
|
||||
session.state = AudioSessionState.CLOSED;
|
||||
sendState(session, "grpcCompleted");
|
||||
}
|
||||
sessionMap.remove(sessionId);
|
||||
removeWebSocketMapping(session);
|
||||
}
|
||||
|
||||
private void failSession(AudioSession session, String reason, boolean cancelRobotStream) {
|
||||
sessionMap.remove(session.sessionId, session);
|
||||
session.state = AudioSessionState.FAILED;
|
||||
ObjectNode event = baseMessage("error", session.sessionId);
|
||||
event.put("message", reason == null ? "未知错误" : reason);
|
||||
event.put("state", session.state.name());
|
||||
sendJson(session.webSocketSession, event);
|
||||
if (cancelRobotStream && session.robotStream != null) {
|
||||
try {
|
||||
session.robotStream.cancel(reason == null ? "audio session failed" : reason);
|
||||
} catch (Exception e) {
|
||||
log.debug("取消机器人音频流异常，sessionId={}", session.sessionId, e);
|
||||
}
|
||||
}
|
||||
removeWebSocketMapping(session);
|
||||
}
|
||||
|
||||
private void sendState(AudioSession session, String type) {
|
||||
ObjectNode event = baseMessage(type, session.sessionId);
|
||||
event.put("terminalId", session.terminalId);
|
||||
event.put("deviceId", session.deviceId);
|
||||
event.put("state", session.state.name());
|
||||
event.put("createdAt", session.createdAt);
|
||||
event.put("lastRobotHeartbeatAt", session.lastRobotHeartbeatAt);
|
||||
sendJson(session.webSocketSession, event);
|
||||
}
|
||||
|
||||
private void sendError(WebSocketSession session, String sessionId, String message) {
|
||||
ObjectNode event = baseMessage("error", sessionId);
|
||||
event.put("message", message == null ? "未知错误" : message);
|
||||
sendJson(session, event);
|
||||
}
|
||||
|
||||
private ObjectNode baseMessage(String type, String sessionId) {
|
||||
ObjectNode event = objectMapper.createObjectNode();
|
||||
event.put("type", type);
|
||||
if (sessionId != null) {
|
||||
event.put("sessionId", sessionId);
|
||||
}
|
||||
event.put("timestamp", System.currentTimeMillis());
|
||||
return event;
|
||||
}
|
||||
|
||||
private void sendJson(WebSocketSession session, ObjectNode event) {
|
||||
sendText(session, toJson(event));
|
||||
}
|
||||
|
||||
private void sendBinary(WebSocketSession session, byte[] pcm) {
|
||||
if (session == null || !session.isOpen() || pcm == null || pcm.length == 0) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
synchronized (session) {
|
||||
session.sendMessage(new BinaryMessage(pcm));
|
||||
}
|
||||
} catch (IOException e) {
|
||||
log.error("发送WebSocket二进制音频失败，wsSessionId={}", session.getId(), e);
|
||||
}
|
||||
}
|
||||
|
||||
private void sendText(WebSocketSession session, String text) {
|
||||
if (session == null || !session.isOpen()) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
synchronized (session) {
|
||||
session.sendMessage(new TextMessage(text));
|
||||
}
|
||||
} catch (IOException e) {
|
||||
log.error("发送WebSocket信令失败", e);
|
||||
}
|
||||
}
|
||||
|
||||
private String toJson(ObjectNode event) {
|
||||
try {
|
||||
return objectMapper.writeValueAsString(event);
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException("序列化WebSocket信令失败", e);
|
||||
}
|
||||
}
|
||||
|
||||
private String findSessionIdByWebSocket(WebSocketSession webSocketSession) {
|
||||
if (webSocketSession == null) {
|
||||
return null;
|
||||
}
|
||||
String wsId = webSocketSession.getId();
|
||||
for (AudioSession session : sessionMap.values()) {
|
||||
if (session.webSocketSession != null && wsId.equals(session.webSocketSession.getId())) {
|
||||
webSocketSessionMap.put(wsId, session.sessionId);
|
||||
return session.sessionId;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private void removeWebSocketMapping(AudioSession session) {
|
||||
if (session != null && session.webSocketSession != null) {
|
||||
webSocketSessionMap.remove(session.webSocketSession.getId(), session.sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
private String getText(JsonNode node, String fieldName) {
|
||||
JsonNode value = node == null ? null : node.get(fieldName);
|
||||
if (value == null || value.isNull()) {
|
||||
return null;
|
||||
}
|
||||
String text = value.asText();
|
||||
return text == null || text.trim().isEmpty() ? null : text.trim();
|
||||
}
|
||||
|
||||
private String getSdp(JsonNode jsonNode) {
|
||||
String direct = getText(jsonNode, "sdp");
|
||||
if (direct != null) {
|
||||
return direct;
|
||||
}
|
||||
JsonNode description = jsonNode == null ? null : jsonNode.get("description");
|
||||
return getText(description, "sdp");
|
||||
}
|
||||
|
||||
private IceCandidateValue getIceCandidate(JsonNode jsonNode) {
|
||||
JsonNode candidateNode = jsonNode == null ? null : jsonNode.get("candidate");
|
||||
if (candidateNode == null || candidateNode.isTextual()) {
|
||||
IceCandidateValue value = new IceCandidateValue();
|
||||
value.candidate = candidateNode == null ? getText(jsonNode, "candidate") : candidateNode.asText();
|
||||
value.sdpMid = getText(jsonNode, "sdpMid");
|
||||
value.sdpMLineIndex = getInt(jsonNode, "sdpMLineIndex", 0);
|
||||
return value;
|
||||
}
|
||||
|
||||
IceCandidateValue value = new IceCandidateValue();
|
||||
value.candidate = getText(candidateNode, "candidate");
|
||||
value.sdpMid = getText(candidateNode, "sdpMid");
|
||||
value.sdpMLineIndex = getInt(candidateNode, "sdpMLineIndex", 0);
|
||||
return value;
|
||||
}
|
||||
|
||||
private int getInt(JsonNode node, String fieldName, int defaultValue) {
|
||||
JsonNode value = node == null ? null : node.get(fieldName);
|
||||
return value == null || !value.isNumber() ? defaultValue : value.asInt(defaultValue);
|
||||
}
|
||||
|
||||
private enum AudioSessionState {
|
||||
CONNECTING,
|
||||
RUNNING,
|
||||
ROBOT_OFFLINE,
|
||||
CLOSED,
|
||||
FAILED
|
||||
}
|
||||
|
||||
/**
|
||||
* 单个实时对讲会话上下文。
|
||||
*/
|
||||
private static class AudioSession {
|
||||
private String sessionId;
|
||||
private String terminalId;
|
||||
private String deviceId;
|
||||
private String operatorId;
|
||||
private volatile WebSocketSession webSocketSession;
|
||||
private volatile RobotAudioStream robotStream;
|
||||
private volatile Consumer<byte[]> robotAudioConsumer;
|
||||
private volatile AudioSessionState state;
|
||||
private volatile long createdAt;
|
||||
private volatile long lastBrowserSignalAt;
|
||||
private volatile long lastBrowserAudioAt;
|
||||
private volatile long lastRobotAudioAt;
|
||||
private volatile long lastRobotHeartbeatAt;
|
||||
}
|
||||
|
||||
private static class IceCandidateValue {
|
||||
private String candidate;
|
||||
private String sdpMid;
|
||||
private int sdpMLineIndex;
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,280 @@
|
||||
package cmvr.api;
|
||||
|
||||
import static io.grpc.MethodDescriptor.generateFullMethodName;
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Real-time bidirectional robot audio service.
|
||||
* Java backend is the gRPC client; robot terminal is the gRPC server.
|
||||
* </pre>
|
||||
*/
|
||||
@javax.annotation.Generated(
|
||||
value = "by gRPC proto compiler (version 1.52.0)",
|
||||
comments = "Source: cmvr/api/robot_audio.proto")
|
||||
@io.grpc.stub.annotations.GrpcGenerated
|
||||
public final class RobotAudioServiceGrpc {
|
||||
|
||||
private RobotAudioServiceGrpc() {}
|
||||
|
||||
public static final String SERVICE_NAME = "cmvr.api.RobotAudioService";
|
||||
|
||||
// Static method descriptors that strictly reflect the proto.
|
||||
private static volatile io.grpc.MethodDescriptor<cmvr.api.RobotAudio.RobotAudioRequest,
|
||||
cmvr.api.RobotAudio.RobotAudioResponse> getAudioTalkMethod;
|
||||
|
||||
@io.grpc.stub.annotations.RpcMethod(
|
||||
fullMethodName = SERVICE_NAME + '/' + "AudioTalk",
|
||||
requestType = cmvr.api.RobotAudio.RobotAudioRequest.class,
|
||||
responseType = cmvr.api.RobotAudio.RobotAudioResponse.class,
|
||||
methodType = io.grpc.MethodDescriptor.MethodType.BIDI_STREAMING)
|
||||
public static io.grpc.MethodDescriptor<cmvr.api.RobotAudio.RobotAudioRequest,
|
||||
cmvr.api.RobotAudio.RobotAudioResponse> getAudioTalkMethod() {
|
||||
io.grpc.MethodDescriptor<cmvr.api.RobotAudio.RobotAudioRequest, cmvr.api.RobotAudio.RobotAudioResponse> getAudioTalkMethod;
|
||||
if ((getAudioTalkMethod = RobotAudioServiceGrpc.getAudioTalkMethod) == null) {
|
||||
synchronized (RobotAudioServiceGrpc.class) {
|
||||
if ((getAudioTalkMethod = RobotAudioServiceGrpc.getAudioTalkMethod) == null) {
|
||||
RobotAudioServiceGrpc.getAudioTalkMethod = getAudioTalkMethod =
|
||||
io.grpc.MethodDescriptor.<cmvr.api.RobotAudio.RobotAudioRequest, cmvr.api.RobotAudio.RobotAudioResponse>newBuilder()
|
||||
.setType(io.grpc.MethodDescriptor.MethodType.BIDI_STREAMING)
|
||||
.setFullMethodName(generateFullMethodName(SERVICE_NAME, "AudioTalk"))
|
||||
.setSampledToLocalTracing(true)
|
||||
.setRequestMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
|
||||
cmvr.api.RobotAudio.RobotAudioRequest.getDefaultInstance()))
|
||||
.setResponseMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
|
||||
cmvr.api.RobotAudio.RobotAudioResponse.getDefaultInstance()))
|
||||
.setSchemaDescriptor(new RobotAudioServiceMethodDescriptorSupplier("AudioTalk"))
|
||||
.build();
|
||||
}
|
||||
}
|
||||
}
|
||||
return getAudioTalkMethod;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new async stub that supports all call types for the service
|
||||
*/
|
||||
public static RobotAudioServiceStub newStub(io.grpc.Channel channel) {
|
||||
io.grpc.stub.AbstractStub.StubFactory<RobotAudioServiceStub> factory =
|
||||
new io.grpc.stub.AbstractStub.StubFactory<RobotAudioServiceStub>() {
|
||||
@java.lang.Override
|
||||
public RobotAudioServiceStub newStub(io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
|
||||
return new RobotAudioServiceStub(channel, callOptions);
|
||||
}
|
||||
};
|
||||
return RobotAudioServiceStub.newStub(factory, channel);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new blocking-style stub that supports unary and streaming output calls on the service
|
||||
*/
|
||||
public static RobotAudioServiceBlockingStub newBlockingStub(
|
||||
io.grpc.Channel channel) {
|
||||
io.grpc.stub.AbstractStub.StubFactory<RobotAudioServiceBlockingStub> factory =
|
||||
new io.grpc.stub.AbstractStub.StubFactory<RobotAudioServiceBlockingStub>() {
|
||||
@java.lang.Override
|
||||
public RobotAudioServiceBlockingStub newStub(io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
|
||||
return new RobotAudioServiceBlockingStub(channel, callOptions);
|
||||
}
|
||||
};
|
||||
return RobotAudioServiceBlockingStub.newStub(factory, channel);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new ListenableFuture-style stub that supports unary calls on the service
|
||||
*/
|
||||
public static RobotAudioServiceFutureStub newFutureStub(
|
||||
io.grpc.Channel channel) {
|
||||
io.grpc.stub.AbstractStub.StubFactory<RobotAudioServiceFutureStub> factory =
|
||||
new io.grpc.stub.AbstractStub.StubFactory<RobotAudioServiceFutureStub>() {
|
||||
@java.lang.Override
|
||||
public RobotAudioServiceFutureStub newStub(io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
|
||||
return new RobotAudioServiceFutureStub(channel, callOptions);
|
||||
}
|
||||
};
|
||||
return RobotAudioServiceFutureStub.newStub(factory, channel);
|
||||
}
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Real-time bidirectional robot audio service.
|
||||
* Java backend is the gRPC client; robot terminal is the gRPC server.
|
||||
* </pre>
|
||||
*/
|
||||
public static abstract class RobotAudioServiceImplBase implements io.grpc.BindableService {
|
||||
|
||||
/**
|
||||
*/
|
||||
public io.grpc.stub.StreamObserver<cmvr.api.RobotAudio.RobotAudioRequest> audioTalk(
|
||||
io.grpc.stub.StreamObserver<cmvr.api.RobotAudio.RobotAudioResponse> responseObserver) {
|
||||
return io.grpc.stub.ServerCalls.asyncUnimplementedStreamingCall(getAudioTalkMethod(), responseObserver);
|
||||
}
|
||||
|
||||
@java.lang.Override public final io.grpc.ServerServiceDefinition bindService() {
|
||||
return io.grpc.ServerServiceDefinition.builder(getServiceDescriptor())
|
||||
.addMethod(
|
||||
getAudioTalkMethod(),
|
||||
io.grpc.stub.ServerCalls.asyncBidiStreamingCall(
|
||||
new MethodHandlers<
|
||||
cmvr.api.RobotAudio.RobotAudioRequest,
|
||||
cmvr.api.RobotAudio.RobotAudioResponse>(
|
||||
this, METHODID_AUDIO_TALK)))
|
||||
.build();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Real-time bidirectional robot audio service.
|
||||
* Java backend is the gRPC client; robot terminal is the gRPC server.
|
||||
* </pre>
|
||||
*/
|
||||
public static final class RobotAudioServiceStub extends io.grpc.stub.AbstractAsyncStub<RobotAudioServiceStub> {
|
||||
private RobotAudioServiceStub(
|
||||
io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
|
||||
super(channel, callOptions);
|
||||
}
|
||||
|
||||
@java.lang.Override
|
||||
protected RobotAudioServiceStub build(
|
||||
io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
|
||||
return new RobotAudioServiceStub(channel, callOptions);
|
||||
}
|
||||
|
||||
/**
|
||||
*/
|
||||
public io.grpc.stub.StreamObserver<cmvr.api.RobotAudio.RobotAudioRequest> audioTalk(
|
||||
io.grpc.stub.StreamObserver<cmvr.api.RobotAudio.RobotAudioResponse> responseObserver) {
|
||||
return io.grpc.stub.ClientCalls.asyncBidiStreamingCall(
|
||||
getChannel().newCall(getAudioTalkMethod(), getCallOptions()), responseObserver);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Real-time bidirectional robot audio service.
|
||||
* Java backend is the gRPC client; robot terminal is the gRPC server.
|
||||
* </pre>
|
||||
*/
|
||||
public static final class RobotAudioServiceBlockingStub extends io.grpc.stub.AbstractBlockingStub<RobotAudioServiceBlockingStub> {
|
||||
private RobotAudioServiceBlockingStub(
|
||||
io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
|
||||
super(channel, callOptions);
|
||||
}
|
||||
|
||||
@java.lang.Override
|
||||
protected RobotAudioServiceBlockingStub build(
|
||||
io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
|
||||
return new RobotAudioServiceBlockingStub(channel, callOptions);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Real-time bidirectional robot audio service.
|
||||
* Java backend is the gRPC client; robot terminal is the gRPC server.
|
||||
* </pre>
|
||||
*/
|
||||
public static final class RobotAudioServiceFutureStub extends io.grpc.stub.AbstractFutureStub<RobotAudioServiceFutureStub> {
|
||||
private RobotAudioServiceFutureStub(
|
||||
io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
|
||||
super(channel, callOptions);
|
||||
}
|
||||
|
||||
@java.lang.Override
|
||||
protected RobotAudioServiceFutureStub build(
|
||||
io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
|
||||
return new RobotAudioServiceFutureStub(channel, callOptions);
|
||||
}
|
||||
}
|
||||
|
||||
private static final int METHODID_AUDIO_TALK = 0;
|
||||
|
||||
private static final class MethodHandlers<Req, Resp> implements
|
||||
io.grpc.stub.ServerCalls.UnaryMethod<Req, Resp>,
|
||||
io.grpc.stub.ServerCalls.ServerStreamingMethod<Req, Resp>,
|
||||
io.grpc.stub.ServerCalls.ClientStreamingMethod<Req, Resp>,
|
||||
io.grpc.stub.ServerCalls.BidiStreamingMethod<Req, Resp> {
|
||||
private final RobotAudioServiceImplBase serviceImpl;
|
||||
private final int methodId;
|
||||
|
||||
MethodHandlers(RobotAudioServiceImplBase serviceImpl, int methodId) {
|
||||
this.serviceImpl = serviceImpl;
|
||||
this.methodId = methodId;
|
||||
}
|
||||
|
||||
@java.lang.Override
|
||||
@java.lang.SuppressWarnings("unchecked")
|
||||
public void invoke(Req request, io.grpc.stub.StreamObserver<Resp> responseObserver) {
|
||||
switch (methodId) {
|
||||
default:
|
||||
throw new AssertionError();
|
||||
}
|
||||
}
|
||||
|
||||
@java.lang.Override
|
||||
@java.lang.SuppressWarnings("unchecked")
|
||||
public io.grpc.stub.StreamObserver<Req> invoke(
|
||||
io.grpc.stub.StreamObserver<Resp> responseObserver) {
|
||||
switch (methodId) {
|
||||
case METHODID_AUDIO_TALK:
|
||||
return (io.grpc.stub.StreamObserver<Req>) serviceImpl.audioTalk(
|
||||
(io.grpc.stub.StreamObserver<cmvr.api.RobotAudio.RobotAudioResponse>) responseObserver);
|
||||
default:
|
||||
throw new AssertionError();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static abstract class RobotAudioServiceBaseDescriptorSupplier
|
||||
implements io.grpc.protobuf.ProtoFileDescriptorSupplier, io.grpc.protobuf.ProtoServiceDescriptorSupplier {
|
||||
RobotAudioServiceBaseDescriptorSupplier() {}
|
||||
|
||||
@java.lang.Override
|
||||
public com.google.protobuf.Descriptors.FileDescriptor getFileDescriptor() {
|
||||
return cmvr.api.RobotAudio.getDescriptor();
|
||||
}
|
||||
|
||||
@java.lang.Override
|
||||
public com.google.protobuf.Descriptors.ServiceDescriptor getServiceDescriptor() {
|
||||
return getFileDescriptor().findServiceByName("RobotAudioService");
|
||||
}
|
||||
}
|
||||
|
||||
private static final class RobotAudioServiceFileDescriptorSupplier
|
||||
extends RobotAudioServiceBaseDescriptorSupplier {
|
||||
RobotAudioServiceFileDescriptorSupplier() {}
|
||||
}
|
||||
|
||||
private static final class RobotAudioServiceMethodDescriptorSupplier
|
||||
extends RobotAudioServiceBaseDescriptorSupplier
|
||||
implements io.grpc.protobuf.ProtoMethodDescriptorSupplier {
|
||||
private final String methodName;
|
||||
|
||||
RobotAudioServiceMethodDescriptorSupplier(String methodName) {
|
||||
this.methodName = methodName;
|
||||
}
|
||||
|
||||
@java.lang.Override
|
||||
public com.google.protobuf.Descriptors.MethodDescriptor getMethodDescriptor() {
|
||||
return getServiceDescriptor().findMethodByName(methodName);
|
||||
}
|
||||
}
|
||||
|
||||
private static volatile io.grpc.ServiceDescriptor serviceDescriptor;
|
||||
|
||||
public static io.grpc.ServiceDescriptor getServiceDescriptor() {
|
||||
io.grpc.ServiceDescriptor result = serviceDescriptor;
|
||||
if (result == null) {
|
||||
synchronized (RobotAudioServiceGrpc.class) {
|
||||
result = serviceDescriptor;
|
||||
if (result == null) {
|
||||
serviceDescriptor = result = io.grpc.ServiceDescriptor.newBuilder(SERVICE_NAME)
|
||||
.setSchemaDescriptor(new RobotAudioServiceFileDescriptorSupplier())
|
||||
.addMethod(getAudioTalkMethod())
|
||||
.build();
|
||||
}
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,81 @@
|
||||
syntax = "proto3";
|
||||
|
||||
import "cmvr/api/common.proto";
|
||||
|
||||
package cmvr.api;
|
||||
|
||||
// Real-time bidirectional robot audio service.
|
||||
// Java backend is the gRPC client; robot terminal is the gRPC server.
|
||||
service RobotAudioService {
|
||||
rpc AudioTalk(stream RobotAudioRequest) returns (stream RobotAudioResponse);
|
||||
}
|
||||
|
||||
// Fixed audio format: PCM_S16LE, 48 kHz, 16-bit, mono.
|
||||
// Both directions use 10 ms frames: 480 samples, 960 bytes.
|
||||
message RobotAudioFormat {
|
||||
int32 sample_rate = 1;
|
||||
int32 channels = 2;
|
||||
int32 bits_per_sample = 3;
|
||||
string encoding = 4;
|
||||
}
|
||||
|
||||
message RobotAudioStart {
|
||||
string session_id = 1;
|
||||
string operator_id = 2;
|
||||
RobotAudioFormat format = 3;
|
||||
}
|
||||
|
||||
message RobotAudioFrame {
|
||||
string session_id = 1;
|
||||
int64 sequence = 2;
|
||||
// Robot upload: capture timestamp. Browser downlink: backend send timestamp.
|
||||
int64 timestamp_ms = 3;
|
||||
// PCM_S16LE. Fixed 10 ms / 960 bytes at 48 kHz mono 16-bit.
|
||||
bytes pcm = 4;
|
||||
}
|
||||
|
||||
message RobotAudioHeartbeat {
|
||||
string session_id = 1;
|
||||
int64 timestamp_ms = 2;
|
||||
}
|
||||
|
||||
message RobotAudioStop {
|
||||
string session_id = 1;
|
||||
string reason = 2;
|
||||
}
|
||||
|
||||
message RobotAudioError {
|
||||
string session_id = 1;
|
||||
int32 code = 2;
|
||||
string message = 3;
|
||||
}
|
||||
|
||||
message RobotAudioStartAck {
|
||||
string session_id = 1;
|
||||
bool accepted = 2;
|
||||
string message = 3;
|
||||
}
|
||||
|
||||
message RobotAudioRequest {
|
||||
CommandHeader.Request header = 1;
|
||||
|
||||
oneof payload {
|
||||
RobotAudioStart start = 2;
|
||||
RobotAudioFrame audio_frame = 3;
|
||||
RobotAudioHeartbeat heartbeat = 4;
|
||||
RobotAudioStop stop = 5;
|
||||
RobotAudioError error = 6;
|
||||
}
|
||||
}
|
||||
|
||||
message RobotAudioResponse {
|
||||
CommandHeader.Feedback header = 1;
|
||||
|
||||
oneof payload {
|
||||
RobotAudioStartAck start_ack = 2;
|
||||
RobotAudioFrame audio_frame = 3;
|
||||
RobotAudioHeartbeat heartbeat = 4;
|
||||
RobotAudioStop stop = 5;
|
||||
RobotAudioError error = 6;
|
||||
}
|
||||
}
|
||||
BIN
logs/audio-dump/robot-to-backend-1784086438864.wav
Normal file
BIN
logs/audio-dump/robot-to-backend-1784086438864.wav
Normal file
Binary file not shown.
BIN
logs/audio-dump/robot-to-backend-1784086792085.wav
Normal file
BIN
logs/audio-dump/robot-to-backend-1784086792085.wav
Normal file
Binary file not shown.
Loading…
Reference in New Issue
Block a user