AsyncBatch.java

/*
 * A folder of long PDFs, submitted to the queue and collected as they finish.
 *
 * Two things make this different from a loop of analyze() calls:
 *
 *   1. Nothing is held open for 60 seconds. A 50-page contract can take minutes, and a
 *      synchronous request would be killed at 60 s with a 504 sync_timeout.
 *   2. A stable idempotency key per file means re-running after a crash replays the stored
 *      responses instead of paying for the batch twice.
 *
 *   java -cp target/visionapi-java-1.0.0.jar examples/AsyncBatch.java ./contracts 4
 */

import io.visionapi.AnalyzeRequest;
import io.visionapi.ApiException;
import io.visionapi.Task;
import io.visionapi.TaskFailedException;
import io.visionapi.VisionApi;
import io.visionapi.WaitOptions;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
import java.time.LocalDate;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

public class AsyncBatch {

    public static void main(String[] args) throws IOException, InterruptedException {
        Path folder = Path.of(args.length > 0 ? args[0] : "./contracts");
        int workers = args.length > 1 ? Integer.parseInt(args[1]) : 4;

        VisionApi vision = VisionApi.create();
        String batch = LocalDate.now().toString(); // one idempotency namespace per day

        List<Path> files;
        try (Stream<Path> stream = Files.list(folder)) {
            files = stream.filter(p -> p.toString().endsWith(".pdf")).sorted().toList();
        }

        AtomicInteger spent = new AtomicInteger();

        // The rate limit is 60 requests/minute per key, and polling counts. Four concurrent
        // files at one poll every 2 s is comfortably inside it.
        ExecutorService pool = Executors.newFixedThreadPool(workers);
        for (Path path : files) {
            pool.submit(() -> {
                String name = path.getFileName().toString();
                try {
                    var ref = vision.analyzeAsync(AnalyzeRequest.builder()
                            .file(path)
                            .preset("contract")
                            .pages("1-50")
                            // Derived from the file, not random: a re-run of the same batch
                            // must not re-charge.
                            .idempotencyKey(batch + ":" + name)
                            .build());

                    Task task = vision.waitForTask(ref.taskId(), WaitOptions.defaults()
                            .pollInterval(Duration.ofSeconds(2))
                            .maxWait(Duration.ofMinutes(20)));

                    spent.addAndGet(task.creditsUsed());
                    System.out.println("✓ " + name + ": " + task.creditsUsed() + " credits");
                } catch (TaskFailedException e) {
                    // A failed task costs 0 credits — the reservation is released in full.
                    System.out.println("✗ " + name + ": " + e.code());
                } catch (ApiException e) {
                    System.out.println("✗ " + name + ": " + e.code());
                }
            });
        }

        pool.shutdown();
        pool.awaitTermination(1, TimeUnit.HOURS);
        System.out.printf("%n%d files, %d credits%n", files.size(), spent.get());
    }
}