使用Java AWS SDK时，如何向StartCallAnalyticsStreamTranscriptionRequest发送ConfigurationEvent

问题描述 #

调用StartCallAnalyticsStreamTranscription时，尝试将ConfigurationEvent通过请求的overrideConfiguration传入，抛出以下异常：

Caused by: software.amazon.awssdk.services.transcribestreaming.model.BadRequestException: You must provide ConfigurationEvent as the first event in the stream (Service: transcribe, Status Code: 400, Request ID: e66c528c-55ae-4bf4-a0da-aa2d389ec1cf)

错误原因 #

ConfigurationEvent不属于请求参数的一部分，而是事件流的组成部分。根据AWS Transcribe Streaming API要求，它必须是发送给服务的第一个事件，用来配置通道角色等分析规则，而非通过请求执行属性传递。

解决方案 #

1. 调整事件流发送逻辑，优先发送ConfigurationEvent #

修改SubscriptionImpl类，添加标记确保首次请求时先发送ConfigurationEvent，再发送音频事件：

public static class SubscriptionImpl implements Subscription {
    private static final int CHUNK_SIZE_IN_BYTES = 1024;
    private final Subscriber<? super AudioStream> subscriber;
    private final InputStream inputStream;
    private final ExecutorService executor = Executors.newFixedThreadPool(1);
    private final AtomicLong demand = new AtomicLong(0);
    private final ConfigurationEvent configurationEvent;
    private final AtomicBoolean sentConfigurationEvent = new AtomicBoolean(false);

    // 构造方法新增configurationEvent参数
    SubscriptionImpl(Subscriber<? super AudioStream> s, InputStream inputStream, ConfigurationEvent configurationEvent) {
        this.subscriber = s;
        this.inputStream = inputStream;
        this.configurationEvent = configurationEvent;
    }

    @Override
    public void request(long n) {
        if (n <= 0) {
            subscriber.onError(new IllegalArgumentException("Demand must be positive"));
        }
        demand.getAndAdd(n);
        executor.submit(() -> {
            try {
                // 首次请求时发送ConfigurationEvent作为第一个事件
                if (sentConfigurationEvent.compareAndSet(false, true)) {
                    subscriber.onNext(configurationEvent);
                    demand.decrementAndGet();
                }

                do {
                    ByteBuffer audioBuffer = getNextEvent();
                    if (audioBuffer.remaining() > 0) {
                        AudioEvent audioEvent = audioEventFromBuffer(audioBuffer);
                        subscriber.onNext(audioEvent);
                    } else {
                        subscriber.onComplete();
                        break;
                    }
                } while (demand.decrementAndGet() > 0);
            } catch (Exception e) {
                subscriber.onError(e);
            }
        });
    }

    // 保留原有getNextEvent、audioEventFromBuffer、cancel方法
}

2. 修改CallAnalyticsAudioStreamPublisher，传递ConfigurationEvent到SubscriptionImpl #

private static class CallAnalyticsAudioStreamPublisher implements Publisher<AudioStream> {
    private final ConfigurationEvent configurationEvent;
    private final InputStream inputStream;
    private static Subscription currentSubscription;

    private CallAnalyticsAudioStreamPublisher(String fileName, InputStream inputStream, ConfigurationEvent configurationEvent) {
        this.configurationEvent = configurationEvent;
        this.inputStream = inputStream;
    }

    @Override
    public void subscribe(Subscriber<? super AudioStream> s) {
        if (currentSubscription == null) {
            // 传入configurationEvent创建SubscriptionImpl
            currentSubscription = new SubscriptionImpl(s, inputStream, configurationEvent);
        } else {
            currentSubscription.cancel();
            currentSubscription = new SubscriptionImpl(s, inputStream, configurationEvent);
        }
        s.onSubscribe(currentSubscription);
    }
}

3. 移除请求中的错误配置 #

修改getAnalyticsRequest方法，删除overrideConfiguration中关于ConfigurationEvent的设置：

private static StartCallAnalyticsStreamTranscriptionRequest getAnalyticsRequest(File inputFile) throws IOException, UnsupportedAudioFileException {
    AudioInputStream audioInputStream = AudioSystem.getAudioInputStream(inputFile);
    AudioFormat audioFormat = audioInputStream.getFormat();
    return StartCallAnalyticsStreamTranscriptionRequest.builder()
            .mediaSampleRateHertz(8000)
            .enablePartialResultsStabilization(true)
            .partialResultsStability("high")
            .languageCode(CallAnalyticsLanguageCode.EN_US)
            .mediaEncoding(getAwsMediaEncoding(audioFormat))
            // 移除overrideConfiguration中的ConfigurationEvent相关代码
            .build();
}

说明 #

调整后，事件流会先发送ConfigurationEvent告知服务通道配置，再逐块发送音频数据，符合AWS Transcribe Call Analytics的接口要求，即可解决400错误。

内容的提问来源于stack exchange，提问作者Kryptix