Speech recognition best practices

import com.alibaba.dashscope.audio.asr.recognition.Recognition;
import com.alibaba.dashscope.audio.asr.recognition.RecognitionParam;
import com.alibaba.dashscope.audio.asr.recognition.RecognitionResult;
import com.alibaba.dashscope.common.ResultCallback;
import com.alibaba.dashscope.utils.Constants;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import java.io.InputStream;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;

public class RecognitionPoolExample {
    private static final int WORKERS = 3;
    private static final int TASKS = 9;
    private static final int COMPLETION_TIMEOUT_SECONDS = 60;

    public static void main(String[] args) {
        int exitCode = 0;
        try {
            run(args);
        } catch (Exception e) {
            e.printStackTrace();
            exitCode = 1;
        }
        // Exit this standalone example after all owned resources are closed.
        System.exit(exitCode);
    }

    private static void run(String[] args) throws Exception {
        String apiKey = System.getenv("DASHSCOPE_API_KEY");
        if (apiKey == null || apiKey.isEmpty()) {
            throw new IllegalArgumentException("Set DASHSCOPE_API_KEY first");
        }
        // Replace {WorkspaceId} with the workspace ID for the endpoint region.
        Constants.baseWebsocketApiUrl =
                "wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/inference";
        Path audio = Paths.get(args.length == 0 ? "asr-example.pcm" : args[0]);
        if (!Files.isReadable(audio)) {
            throw new IllegalArgumentException("Audio file is not readable: " + audio);
        }
        int capacity = Integer.parseInt(System.getenv().getOrDefault(
                "RECOGNITION_OBJECTPOOL_SIZE", "500"));
        if (capacity <= 0) {
            throw new IllegalArgumentException("Pool size must be positive");
        }
        GenericObjectPoolConfig<Recognition> config = new GenericObjectPoolConfig<>();
        config.setMaxTotal(capacity);
        config.setMaxIdle(capacity);
        config.setBlockWhenExhausted(true);
        config.setMinIdle(0);

        GenericObjectPool<Recognition> pool =
                new GenericObjectPool<>(new RecognitionFactory(), config);
        ExecutorService executor = Executors.newFixedThreadPool(WORKERS);
        ExecutorService completionExecutor = Executors.newFixedThreadPool(WORKERS);
        try {
            List<Callable<Void>> tasks = new ArrayList<>();
            for (int i = 0; i < TASKS; i++) {
                tasks.add(() -> {
                    recognize(pool, audio, apiKey, completionExecutor);
                    return null;
                });
            }
            // Wait for every task, then propagate any task failure.
            List<Future<Void>> results = executor.invokeAll(tasks);
            for (Future<Void> result : results) {
                result.get();
            }
        } finally {
            executor.shutdown();
            completionExecutor.shutdownNow();
            try {
                if (!executor.awaitTermination(1, TimeUnit.MINUTES)) {
                    executor.shutdownNow();
                }
                if (!completionExecutor.awaitTermination(1, TimeUnit.MINUTES)) {
                    completionExecutor.shutdownNow();
                }
            } finally {
                pool.close();
                System.out.printf("Pool closed: created=%d, destroyed=%d, active=%d, idle=%d%n",
                        pool.getCreatedCount(), pool.getDestroyedCount(),
                        pool.getNumActive(), pool.getNumIdle());
            }
        }
    }

    private static void recognize(GenericObjectPool<Recognition> pool,
                                  Path audio, String apiKey,
                                  ExecutorService completionExecutor) throws Exception {
        // Example timeout; adjust it to the application's queueing policy.
        Recognition recognizer = pool.borrowObject(Duration.ofSeconds(30));
        boolean reusable = false;
        AtomicReference<Exception> failure = new AtomicReference<>();
        CountDownLatch completed = new CountDownLatch(1);
        try (InputStream input = Files.newInputStream(audio)) {
            RecognitionParam param = RecognitionParam.builder()
                    .model("qwen-audio-3.1-asr-flash-streaming")
                    .apiKey(apiKey).format("pcm").sampleRate(16000).build();
            ResultCallback<RecognitionResult> callback = new ResultCallback<RecognitionResult>() {
                @Override
                public void onEvent(RecognitionResult result) {
                    if (result.isSentenceEnd() && result.getSentence() != null) {
                        System.out.println(result.getSentence().getText());
                    }
                }

                @Override
                public void onComplete() {
                    completed.countDown();
                }

                @Override
                public void onError(Exception error) {
                    failure.set(error);
                    completed.countDown();
                }
            };
            recognizer.call(param, callback);
            // 100 ms of PCM16 mono audio at 16 kHz is 3,200 bytes.
            byte[] buffer = new byte[3200];
            int count;
            while ((count = input.read(buffer)) != -1) {
                if (failure.get() != null) {
                    throw failure.get();
                }
                recognizer.sendAudioFrame(ByteBuffer.wrap(buffer, 0, count));
                Thread.sleep(100);
                buffer = new byte[3200];
            }
            // stop() blocks in the SDK, so apply the deadline to the stop call itself.
            long deadline = System.nanoTime()
                    + TimeUnit.SECONDS.toNanos(COMPLETION_TIMEOUT_SECONDS);
            Future<?> stopTask = completionExecutor.submit(recognizer::stop);
            try {
                stopTask.get(Math.max(0, deadline - System.nanoTime()), TimeUnit.NANOSECONDS);
                long remaining = Math.max(0, deadline - System.nanoTime());
                if (!completed.await(remaining, TimeUnit.NANOSECONDS)) {
                    throw new TimeoutException("Timed out waiting for task completion");
                }
            } finally {
                stopTask.cancel(true);
            }
            if (failure.get() != null) {
                throw failure.get();
            }
            reusable = true;
            System.out.printf("Completed: request_id=%s, first_packet_ms=%d, last_packet_ms=%d%n",
                    recognizer.getLastRequestId(), recognizer.getFirstPackageDelay(),
                    recognizer.getLastPackageDelay());
        } finally {
            if (reusable) {
                pool.returnObject(recognizer);
            } else {
                // destroyObject closes the connection before discarding the object.
                pool.invalidateObject(recognizer);
            }
        }
    }

    private static class RecognitionFactory extends BasePooledObjectFactory<Recognition> {
        @Override
        public Recognition create() {
            return new Recognition();
        }

        @Override
        public PooledObject<Recognition> wrap(Recognition recognizer) {
            return new DefaultPooledObject<>(recognizer);
        }

        @Override
        public void destroyObject(PooledObject<Recognition> object) {
            if (object.getObject().getDuplexApi() != null) {
                object.getObject().getDuplexApi().close(1000, "pool cleanup");
            }
        }
    }
}