feat: streaming APIExecutor (#296)

* feat: streaming raw requests

feat: docs e.g. and changelog

fix: changelog

feat: cleanup

feat: address comments

feat: provide typereference overload example

feat: nullable consumer

feat: cleanup

fix: async contract leak

feat: refactor

* feat: address coderabbit comments

* feat: update changelog

* feat: address comments

Author: SoulPancake

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## [Unreleased](https://github.com/openfga/java-sdk/compare/v0.9.6...HEAD)
 
+### Added
+- Introduced `StreamingApiExecutor` for executing HTTP requests to streaming endpoints not yet wrapped by the SDK (#296)
+
 ## v0.9.6
 
 ### [0.9.6](https://github.com/openfga/java-sdk/compare/v0.9.5...v0.9.6) (2026-02-18)

README.md

@@ -1245,11 +1245,33 @@ System.out.println("Status Code: " + response.getStatusCode());
 System.out.println("Headers: " + response.getHeaders());
 ```
 
+#### Calling a streaming endpoint
+
+For streaming endpoints, use `streamingApiExecutor` instead. Pass the response class directly — the SDK handles the rest. It delivers each response object to a consumer callback as it arrives, and returns a `CompletableFuture<Void>` that completes when the stream is exhausted.
+
+```java
+ApiExecutorRequestBuilder request = ApiExecutorRequestBuilder.builder(HttpMethod.POST, "/stores/{store_id}/streamed-list-objects")
+    .body(new ListObjectsRequest().user("user:anne").relation("viewer").type("document"))
+    .build();
+
+fgaClient.streamingApiExecutor(StreamedListObjectsResponse.class)
+    .stream(
+        request,
+        response -> System.out.println("Object: " + response.getObject()),  // called per object
+        error    -> System.err.println("Stream error: " + error.getMessage()) // optional
+    )
+    .thenRun(() -> System.out.println("Streaming complete"))
+    .exceptionally(err -> {
+        System.err.println("Fatal error: " + err.getMessage());
+        return null;
+    });
+```
+
 For a complete working example, see [examples/api-executor](examples/api-executor).
 
 #### Documentation
 
-See [docs/ApiExecutor.md](docs/ApiExecutor.md) for complete API reference and examples.
+See [docs/ApiExecutor.md](docs/ApiExecutor.md) for complete API reference and examples for both `ApiExecutor` and `StreamingApiExecutor`.
 
 ### API Endpoints
 

docs/ApiExecutor.md

@@ -1,29 +1,48 @@
 # API Executor
 
-Direct HTTP access to OpenFGA endpoints.
+Direct HTTP access to OpenFGA endpoints — for both standard and streaming responses.
 
 ## Quick Start
 
+### Standard (non-streaming) endpoint
+
 ```java
 OpenFgaClient client = new OpenFgaClient(config);
 
-// Build request
 ApiExecutorRequestBuilder request = ApiExecutorRequestBuilder.builder(HttpMethod.POST, "/stores/{store_id}/check")
     .pathParam("store_id", storeId)
     .body(Map.of("tuple_key", Map.of("user", "user:jon", "relation", "reader", "object", "doc:1")))
     .build();
 
-// Execute - typed response
+// Typed response
 ApiResponse<CheckResponse> response = client.apiExecutor().send(request, CheckResponse.class).get();
 
-// Execute - raw JSON
+// Raw JSON
 ApiResponse<String> rawResponse = client.apiExecutor().send(request).get();
 ```
 
+### Streaming endpoint
+
+```java
+ApiExecutorRequestBuilder request = ApiExecutorRequestBuilder.builder(HttpMethod.POST, "/stores/{store_id}/streamed-list-objects")
+    .body(listObjectsRequest)
+    .build();
+
+client.streamingApiExecutor(StreamedListObjectsResponse.class)
+    .stream(
+        request,
+        response -> System.out.println("Object: " + response.getObject()),
+        error    -> System.err.println("Error: " + error.getMessage())
+    )
+    .thenRun(() -> System.out.println("Done"));
+```
+
 ## API Reference
 
 ### ApiExecutorRequestBuilder
 
+Shared by both `ApiExecutor` and `StreamingApiExecutor`.
+
 **Factory:**
 ```java
 ApiExecutorRequestBuilder.builder(HttpMethod method, String path)
@@ -38,30 +57,47 @@ ApiExecutorRequestBuilder.builder(HttpMethod method, String path)
 .build()                                  // Complete the builder (required)
 ```
 
-**Example:**
-```java
-ApiExecutorRequestBuilder request = ApiExecutorRequestBuilder.builder(HttpMethod.POST, "/stores/{store_id}/write")
-    .pathParam("store_id", "01ABC")
-    .queryParam("dry_run", "true")
-    .header("X-Request-ID", "uuid")
-    .body(requestObject)
-    .build();
-```
-
 ### ApiExecutor
 
 **Access:**
 ```java
-ApiExecutor apiExecutor = client.apiExecutor();
+ApiExecutor executor = client.apiExecutor();
 ```
 
 **Methods:**
 ```java
 CompletableFuture<ApiResponse<String>> send(ApiExecutorRequestBuilder request)
-CompletableFuture<ApiResponse<T>> send(ApiExecutorRequestBuilder request, Class<T> responseType)
+CompletableFuture<ApiResponse<T>>      send(ApiExecutorRequestBuilder request, Class<T> responseType)
+```
+
+### StreamingApiExecutor\<T\>
+
+For streaming endpoints. Each response object is delivered to a consumer callback as it arrives.
+
+**Access — preferred (concrete response types):**
+```java
+StreamingApiExecutor<MyResponse> executor = client.streamingApiExecutor(MyResponse.class);
 ```
 
-### ApiResponse<T>
+**Access — escape hatch (when T is itself generic):**
+```java
+TypeReference<StreamResult<MyResponse>> typeRef = new TypeReference<StreamResult<MyResponse>>() {};
+StreamingApiExecutor<MyResponse> executor = client.streamingApiExecutor(typeRef);
+```
+
+**Methods:**
+```java
+CompletableFuture<Void> stream(ApiExecutorRequestBuilder request, Consumer<T> consumer)
+CompletableFuture<Void> stream(ApiExecutorRequestBuilder request, Consumer<T> consumer, Consumer<Throwable> errorConsumer)
+```
+
+- The `consumer` is invoked once per successfully parsed response object.
+- The optional `errorConsumer` is invoked for errors within the stream or on HTTP error.
+- The returned `CompletableFuture<Void>` completes when the stream is exhausted or fails exceptionally on unrecoverable error.
+
+### ApiResponse\<T\>
+
+Returned by `ApiExecutor.send(...)`.
 
 ```java
 int getStatusCode()                    // HTTP status
@@ -72,7 +108,7 @@ T getData()                            // Deserialized data
 
 ## Examples
 
-### GET Request
+### GET Request (ApiExecutor)
 ```java
 ApiExecutorRequestBuilder request = ApiExecutorRequestBuilder.builder(HttpMethod.GET, "/stores/{store_id}/feature")
     .pathParam("store_id", storeId)
@@ -82,7 +118,7 @@ client.apiExecutor().send(request, FeatureResponse.class)
     .thenAccept(r -> System.out.println("Status: " + r.getStatusCode()));
 ```
 
-### POST with Body
+### POST with Body (ApiExecutor)
 ```java
 ApiExecutorRequestBuilder request = ApiExecutorRequestBuilder.builder(HttpMethod.POST, "/stores/{store_id}/bulk-delete")
     .pathParam("store_id", storeId)
@@ -93,15 +129,45 @@ ApiExecutorRequestBuilder request = ApiExecutorRequestBuilder.builder(HttpMethod
 client.apiExecutor().send(request, BulkDeleteResponse.class).get();
 ```
 
-### Raw JSON Response
+### Raw JSON Response (ApiExecutor)
 ```java
 ApiResponse<String> response = client.apiExecutor().send(request).get();
-String json = response.getRawResponse(); // Raw JSON
+String json = response.getRawResponse();
+```
+
+### Streaming endpoint (StreamingApiExecutor)
+```java
+ApiExecutorRequestBuilder request = ApiExecutorRequestBuilder.builder(HttpMethod.POST, "/stores/{store_id}/streamed-list-objects")
+    .body(new ListObjectsRequest().user("user:anne").relation("viewer").type("document"))
+    .build();
+
+List<String> objects = new ArrayList<>();
+client.streamingApiExecutor(StreamedListObjectsResponse.class)
+    .stream(request, response -> objects.add(response.getObject()))
+    .thenRun(() -> System.out.println("Received " + objects.size() + " objects"));
+```
+
+### Streaming endpoint with TypeReference (escape hatch for generic response types)
+
+Use `TypeReference` only when the response type `T` is itself generic. For all concrete
+types — which covers the vast majority of endpoints — use `streamingApiExecutor(MyResponse.class)` instead.
+
+```java
+// Hypothetical endpoint whose response wraps a generic Page<Item>
+TypeReference<StreamResult<Page<Item>>> typeRef = new TypeReference<StreamResult<Page<Item>>>() {};
+
+ApiExecutorRequestBuilder request = ApiExecutorRequestBuilder.builder(HttpMethod.POST, "/stores/{store_id}/streamed-paged-items")
+    .body(requestBody)
+    .build();
+
+client.streamingApiExecutor(typeRef)
+    .stream(request, page -> page.getItems().forEach(System.out::println))
+    .thenRun(() -> System.out.println("Done"));
 ```
 
 ### Query Parameters
 ```java
-ApiExecutorRequestBuilder.builder("GET", "/stores/{store_id}/items")
+ApiExecutorRequestBuilder.builder(HttpMethod.GET, "/stores/{store_id}/items")
     .pathParam("store_id", storeId)
     .queryParam("page", "1")
     .queryParam("limit", "50")
@@ -111,14 +177,14 @@ ApiExecutorRequestBuilder.builder("GET", "/stores/{store_id}/items")
 
 ### Custom Headers
 ```java
-ApiExecutorRequestBuilder.builder("POST", "/stores/{store_id}/action")
+ApiExecutorRequestBuilder.builder(HttpMethod.POST, "/stores/{store_id}/action")
     .header("X-Request-ID", UUID.randomUUID().toString())
     .header("X-Idempotency-Key", "key-123")
     .body(data)
     .build();
 ```
 
-### Error Handling
+### Error Handling (ApiExecutor)
 ```java
 client.apiExecutor().send(request, ResponseType.class)
     .exceptionally(e -> {
@@ -132,7 +198,7 @@ client.apiExecutor().send(request, ResponseType.class)
 
 ### Map as Request Body
 ```java
-ApiExecutorRequestBuilder.builder("POST", "/stores/{store_id}/settings")
+ApiExecutorRequestBuilder.builder(HttpMethod.POST, "/stores/{store_id}/settings")
     .pathParam("store_id", storeId)
     .body(Map.of(
         "setting", "value",
@@ -148,17 +214,18 @@ ApiExecutorRequestBuilder.builder("POST", "/stores/{store_id}/settings")
 - Path/query parameters are URL-encoded automatically
 - Authentication tokens injected from client config
 - `{store_id}` auto-replaced if not provided via `.pathParam()`
+- For `StreamingApiExecutor`, pass the response class directly (`MyResponse.class`). The SDK builds the required Jackson type internally. Use the `TypeReference` overload only when `T` is itself a generic type.
 
 ## Migration to Typed Methods
 
-When SDK adds typed methods for an endpoint, you can migrate from API Executor:
+When the SDK adds typed methods for an endpoint, you can migrate from API Executor:
 
 ```java
 // API Executor
-ApiExecutorRequestBuilder request = ApiExecutorRequestBuilder.builder("POST", "/stores/{store_id}/check")
+ApiExecutorRequestBuilder request = ApiExecutorRequestBuilder.builder(HttpMethod.POST, "/stores/{store_id}/check")
     .body(req)
     .build();
-    
+
 client.apiExecutor().send(request, CheckResponse.class).get();
 
 // Typed SDK (when available)

examples/README.md

@@ -13,6 +13,10 @@ A simple example that creates a store, runs a set of calls against it including
 #### Streaming Examples
 - `streamed-list-objects/` - Demonstrates using the StreamedListObjects API to retrieve large result sets without pagination limits
 
-#### API Executor Examples
-- `api-executor/` - Demonstrates direct HTTP access to OpenFGA endpoints not yet wrapped by the SDK, maintaining SDK configuration (authentication, retries, error handling)
+#### API Executor Examples (`api-executor/`)
+Demonstrates direct HTTP access to OpenFGA endpoints not yet wrapped by the SDK. Standard requests go through the SDK's full configuration (authentication, retries, error handling, telemetry). Streaming requests use direct HTTP streaming — authentication is applied but retries and telemetry are not.
 
+- **`ApiExecutorExample.java`** — standard (non-streaming) endpoints: typed responses, raw JSON, query parameters, custom headers, error handling
+- **`StreamingApiExecutorExample.java`** — streaming endpoints: calls `streamed-list-objects` via `client.streamingApiExecutor(StreamedListObjectsResponse.class).stream(request, consumer)`, demonstrating per-object callbacks and error handling
+
+Run with `make run` (standard) or `make run-streaming` (streaming) from the `api-executor/` directory.

examples/api-executor/Makefile

@@ -1,4 +1,4 @@
-.PHONY: build run run-openfga
+.PHONY: build run run-streaming run-openfga
 all: build
 
 project_name=.
@@ -11,6 +11,9 @@ build:
 run:
 	../../gradlew -P language=$(language) run
 
+run-streaming:
+	../../gradlew -P language=$(language) runStreaming
+
 run-openfga:
 	docker pull docker.io/openfga/openfga:${openfga_version} && \
 		docker run -p 8080:8080 docker.io/openfga/openfga:${openfga_version} run