From 67d0d607792689dada84b5583609a0fb05251674 Mon Sep 17 00:00:00 2001 From: Steve Hawkins Date: Fri, 27 Mar 2026 10:16:35 -0400 Subject: [PATCH 1/5] demonstrates client side streaming Signed-off-by: Steve Hawkins --- .../admin/client/StreamRxInvoker.java | 185 ++++++++++++++++++ .../admin/client/StreamRxInvokerProvider.java | 21 ++ .../spi/ResteasyClientClassicProvider.java | 3 +- 3 files changed, 208 insertions(+), 1 deletion(-) create mode 100644 integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvoker.java create mode 100644 integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvokerProvider.java diff --git a/integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvoker.java b/integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvoker.java new file mode 100644 index 000000000000..25146acd1cec --- /dev/null +++ b/integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvoker.java @@ -0,0 +1,185 @@ +package org.keycloak.admin.client; + +import java.io.IOException; +import java.io.InputStream; +import java.lang.reflect.Proxy; +import java.util.Spliterator; +import java.util.Spliterators; +import java.util.concurrent.ExecutorService; +import java.util.stream.Stream; +import java.util.stream.StreamSupport; + +import com.fasterxml.jackson.databind.MappingIterator; +import com.fasterxml.jackson.databind.json.JsonMapper; + +import jakarta.ws.rs.client.Entity; +import jakarta.ws.rs.client.ResponseProcessingException; +import jakarta.ws.rs.client.RxInvoker; +import jakarta.ws.rs.client.SyncInvoker; +import jakarta.ws.rs.core.GenericType; +import jakarta.ws.rs.core.Response; + +public class StreamRxInvoker implements RxInvoker> { + + private SyncInvoker syncInvoker; + private ExecutorService executorService; + + public StreamRxInvoker(SyncInvoker syncInvoker, ExecutorService executorService) { + this.syncInvoker = syncInvoker; + this.executorService = executorService; + } + + @Override + public Stream get() { + throw new UnsupportedOperationException(); + } + + @Override + public Stream get(Class responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream get(GenericType responseType) { + Response response = this.syncInvoker.get(Response.class); + + if (response.getStatus() != 200) { + response.close(); + throw new AssertionError(); // TODO + } + + InputStream stream = (InputStream)response.getEntity(); + + JsonMapper mapper = new JsonMapper(); // TODO: obtain a better instance + try { + MappingIterator iter = mapper.readerFor(responseType.getRawType()).readValues(stream); + + Stream targetStream = StreamSupport.stream( + Spliterators.spliteratorUnknownSize(iter, Spliterator.ORDERED), + false); + return (Stream) Proxy.newProxyInstance(Stream.class.getClassLoader(), new Class[] {Stream.class}, (proxy, method, args) -> { + if (method.getName().equals("close")) { + response.close(); + } + return method.invoke(targetStream, args); + }); + } catch (IOException e) { + response.close(); + throw new ResponseProcessingException(response, "Error", e); + } + } + + @Override + public Stream put(Entity entity) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream put(Entity entity, Class responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream put(Entity entity, GenericType responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream post(Entity entity) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream post(Entity entity, Class responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream post(Entity entity, GenericType responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream delete() { + throw new UnsupportedOperationException(); + } + + @Override + public Stream delete(Class responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream delete(GenericType responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream head() { + throw new UnsupportedOperationException(); + } + + @Override + public Stream options() { + throw new UnsupportedOperationException(); + } + + @Override + public Stream options(Class responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream options(GenericType responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream trace() { + throw new UnsupportedOperationException(); + } + + @Override + public Stream trace(Class responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream trace(GenericType responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream method(String name) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream method(String name, Class responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream method(String name, GenericType responseType) { + if(!name.equals("GET")) { + throw new UnsupportedOperationException(); + } + return get(responseType); + } + + @Override + public Stream method(String name, Entity entity) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream method(String name, Entity entity, Class responseType) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream method(String name, Entity entity, GenericType responseType) { + throw new UnsupportedOperationException(); + } + +} diff --git a/integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvokerProvider.java b/integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvokerProvider.java new file mode 100644 index 000000000000..a989ba192871 --- /dev/null +++ b/integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvokerProvider.java @@ -0,0 +1,21 @@ +package org.keycloak.admin.client; + +import java.util.concurrent.ExecutorService; +import java.util.stream.Stream; + +import jakarta.ws.rs.client.RxInvokerProvider; +import jakarta.ws.rs.client.SyncInvoker; + +public class StreamRxInvokerProvider implements RxInvokerProvider { + + @Override + public boolean isProviderFor(Class clazz) { + return clazz.equals(Stream.class); + } + + @Override + public StreamRxInvoker getRxInvoker(SyncInvoker syncInvoker, ExecutorService executorService) { + return new StreamRxInvoker(syncInvoker, executorService); + } + +} diff --git a/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/ResteasyClientClassicProvider.java b/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/ResteasyClientClassicProvider.java index 65dedd5e13d6..7954dafe5684 100644 --- a/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/ResteasyClientClassicProvider.java +++ b/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/ResteasyClientClassicProvider.java @@ -25,7 +25,7 @@ import org.keycloak.admin.client.ClientBuilderWrapper; import org.keycloak.admin.client.JacksonProvider; - +import org.keycloak.admin.client.StreamRxInvokerProvider; import org.jboss.resteasy.client.jaxrs.ResteasyWebTarget; /** @@ -42,6 +42,7 @@ public Client newRestEasyClient(Object customJacksonProvider, SSLContext sslCont } else { clientBuilder.register(JacksonProvider.class, 100); } + clientBuilder.register(StreamRxInvokerProvider.class); return clientBuilder.build(); } From 47389ee77b72da4aa6edd5a0698c76890f6086f8 Mon Sep 17 00:00:00 2001 From: Steve Hawkins Date: Mon, 30 Mar 2026 21:50:59 -0400 Subject: [PATCH 2/5] fix: adding client streaming support closes: #47542 Signed-off-by: Steve Hawkins --- .../admin/client/ClientBuilderWrapper.java | 36 ++-- .../admin/client/JacksonProvider.java | 10 +- .../admin/client/KeycloakBuilder.java | 2 +- .../admin/client/StreamRxInvoker.java | 185 ------------------ .../admin/client/StreamRxInvokerProvider.java | 21 -- .../deserializer/StreamDeserializer.java | 55 ------ .../spi/ResteasyClientClassicProvider.java | 25 ++- .../client/spi/StreamMessageBodyReader.java | 81 ++++++++ 8 files changed, 113 insertions(+), 302 deletions(-) delete mode 100644 integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvoker.java delete mode 100644 integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvokerProvider.java delete mode 100644 integration/admin-client/src/main/java/org/keycloak/admin/client/deserializer/StreamDeserializer.java create mode 100644 integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java diff --git a/integration/admin-client/src/main/java/org/keycloak/admin/client/ClientBuilderWrapper.java b/integration/admin-client/src/main/java/org/keycloak/admin/client/ClientBuilderWrapper.java index d183411e881d..b9474d5efc3a 100644 --- a/integration/admin-client/src/main/java/org/keycloak/admin/client/ClientBuilderWrapper.java +++ b/integration/admin-client/src/main/java/org/keycloak/admin/client/ClientBuilderWrapper.java @@ -4,33 +4,23 @@ import jakarta.ws.rs.client.ClientBuilder; -public class ClientBuilderWrapper { +import org.keycloak.admin.client.spi.ResteasyClientClassicProvider; - static Class clazz; - static { - try { - clazz = Class.forName("org.jboss.resteasy.client.jaxrs.internal.ResteasyClientBuilderImpl"); - } catch (ClassNotFoundException e) { - try { - clazz = Class.forName("org.jboss.resteasy.client.jaxrs.ResteasyClientBuilder"); - } catch (ClassNotFoundException ex) { - throw new RuntimeException("RestEasy 3 or 4 not found on classpath"); - } - } - } +import org.jboss.resteasy.client.jaxrs.internal.ResteasyClientBuilderImpl; + +/** + * Use {@link ResteasyClientClassicProvider#createClientBuilder()} instead + */ +@Deprecated +public class ClientBuilderWrapper { public static ClientBuilder create(SSLContext sslContext, boolean disableTrustManager) { - try { - Object o = clazz.newInstance(); - clazz.getMethod("sslContext", SSLContext.class).invoke(o, sslContext); - clazz.getMethod("connectionPoolSize", int.class).invoke(o, 10); - if (disableTrustManager) { - clazz.getMethod("disableTrustManager").invoke(o); - } - return (ClientBuilder) o; - } catch (Exception e) { - throw new RuntimeException(e); + ResteasyClientBuilderImpl result = ResteasyClientClassicProvider.createClientBuilder(); + result.sslContext(sslContext); + if (disableTrustManager) { + result.disableTrustManager(); } + return result; } } diff --git a/integration/admin-client/src/main/java/org/keycloak/admin/client/JacksonProvider.java b/integration/admin-client/src/main/java/org/keycloak/admin/client/JacksonProvider.java index 318ff8860f6f..75a7005e618d 100644 --- a/integration/admin-client/src/main/java/org/keycloak/admin/client/JacksonProvider.java +++ b/integration/admin-client/src/main/java/org/keycloak/admin/client/JacksonProvider.java @@ -1,32 +1,24 @@ package org.keycloak.admin.client; -import java.util.stream.Stream; - import jakarta.ws.rs.core.MediaType; -import org.keycloak.admin.client.deserializer.StreamDeserializer; - import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.module.SimpleModule; import org.jboss.resteasy.plugins.providers.jackson.ResteasyJackson2Provider; public class JacksonProvider extends ResteasyJackson2Provider { - private static final StreamDeserializer STREAM_DESERIALIZER = new StreamDeserializer(); - private static final SimpleModule STREAM_MODULE = new SimpleModule().addDeserializer(Stream.class, STREAM_DESERIALIZER); @Override public ObjectMapper locateMapper(Class type, MediaType mediaType) { ObjectMapper objectMapper = super.locateMapper(type, mediaType); // Same like JSONSerialization class. Makes it possible to use admin-client against older versions of Keycloak server where the properties on representations might be different - objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL); + objectMapper.setDefaultPropertyInclusion(JsonInclude.Include.NON_NULL); // The client must work with the newer versions of Keycloak server, which might contain the JSON fields not yet known by the client. So unknown fields will be ignored. objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); - objectMapper.registerModule(STREAM_MODULE); return objectMapper; } } diff --git a/integration/admin-client/src/main/java/org/keycloak/admin/client/KeycloakBuilder.java b/integration/admin-client/src/main/java/org/keycloak/admin/client/KeycloakBuilder.java index e12a939fb941..42e36ca7cb69 100644 --- a/integration/admin-client/src/main/java/org/keycloak/admin/client/KeycloakBuilder.java +++ b/integration/admin-client/src/main/java/org/keycloak/admin/client/KeycloakBuilder.java @@ -34,7 +34,7 @@ * .password("pass") * .clientId("client") * .clientSecret("secret") - * .resteasyClient(new ResteasyClientBuilderImpl() + * .resteasyClient(ResteasyClientClassicProvider.createClientBuilder() * .connectionPoolSize(20) * .build() * .register(org.keycloak.admin.client.JacksonProvider.class, 100)) diff --git a/integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvoker.java b/integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvoker.java deleted file mode 100644 index 25146acd1cec..000000000000 --- a/integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvoker.java +++ /dev/null @@ -1,185 +0,0 @@ -package org.keycloak.admin.client; - -import java.io.IOException; -import java.io.InputStream; -import java.lang.reflect.Proxy; -import java.util.Spliterator; -import java.util.Spliterators; -import java.util.concurrent.ExecutorService; -import java.util.stream.Stream; -import java.util.stream.StreamSupport; - -import com.fasterxml.jackson.databind.MappingIterator; -import com.fasterxml.jackson.databind.json.JsonMapper; - -import jakarta.ws.rs.client.Entity; -import jakarta.ws.rs.client.ResponseProcessingException; -import jakarta.ws.rs.client.RxInvoker; -import jakarta.ws.rs.client.SyncInvoker; -import jakarta.ws.rs.core.GenericType; -import jakarta.ws.rs.core.Response; - -public class StreamRxInvoker implements RxInvoker> { - - private SyncInvoker syncInvoker; - private ExecutorService executorService; - - public StreamRxInvoker(SyncInvoker syncInvoker, ExecutorService executorService) { - this.syncInvoker = syncInvoker; - this.executorService = executorService; - } - - @Override - public Stream get() { - throw new UnsupportedOperationException(); - } - - @Override - public Stream get(Class responseType) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream get(GenericType responseType) { - Response response = this.syncInvoker.get(Response.class); - - if (response.getStatus() != 200) { - response.close(); - throw new AssertionError(); // TODO - } - - InputStream stream = (InputStream)response.getEntity(); - - JsonMapper mapper = new JsonMapper(); // TODO: obtain a better instance - try { - MappingIterator iter = mapper.readerFor(responseType.getRawType()).readValues(stream); - - Stream targetStream = StreamSupport.stream( - Spliterators.spliteratorUnknownSize(iter, Spliterator.ORDERED), - false); - return (Stream) Proxy.newProxyInstance(Stream.class.getClassLoader(), new Class[] {Stream.class}, (proxy, method, args) -> { - if (method.getName().equals("close")) { - response.close(); - } - return method.invoke(targetStream, args); - }); - } catch (IOException e) { - response.close(); - throw new ResponseProcessingException(response, "Error", e); - } - } - - @Override - public Stream put(Entity entity) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream put(Entity entity, Class responseType) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream put(Entity entity, GenericType responseType) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream post(Entity entity) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream post(Entity entity, Class responseType) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream post(Entity entity, GenericType responseType) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream delete() { - throw new UnsupportedOperationException(); - } - - @Override - public Stream delete(Class responseType) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream delete(GenericType responseType) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream head() { - throw new UnsupportedOperationException(); - } - - @Override - public Stream options() { - throw new UnsupportedOperationException(); - } - - @Override - public Stream options(Class responseType) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream options(GenericType responseType) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream trace() { - throw new UnsupportedOperationException(); - } - - @Override - public Stream trace(Class responseType) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream trace(GenericType responseType) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream method(String name) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream method(String name, Class responseType) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream method(String name, GenericType responseType) { - if(!name.equals("GET")) { - throw new UnsupportedOperationException(); - } - return get(responseType); - } - - @Override - public Stream method(String name, Entity entity) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream method(String name, Entity entity, Class responseType) { - throw new UnsupportedOperationException(); - } - - @Override - public Stream method(String name, Entity entity, GenericType responseType) { - throw new UnsupportedOperationException(); - } - -} diff --git a/integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvokerProvider.java b/integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvokerProvider.java deleted file mode 100644 index a989ba192871..000000000000 --- a/integration/admin-client/src/main/java/org/keycloak/admin/client/StreamRxInvokerProvider.java +++ /dev/null @@ -1,21 +0,0 @@ -package org.keycloak.admin.client; - -import java.util.concurrent.ExecutorService; -import java.util.stream.Stream; - -import jakarta.ws.rs.client.RxInvokerProvider; -import jakarta.ws.rs.client.SyncInvoker; - -public class StreamRxInvokerProvider implements RxInvokerProvider { - - @Override - public boolean isProviderFor(Class clazz) { - return clazz.equals(Stream.class); - } - - @Override - public StreamRxInvoker getRxInvoker(SyncInvoker syncInvoker, ExecutorService executorService) { - return new StreamRxInvoker(syncInvoker, executorService); - } - -} diff --git a/integration/admin-client/src/main/java/org/keycloak/admin/client/deserializer/StreamDeserializer.java b/integration/admin-client/src/main/java/org/keycloak/admin/client/deserializer/StreamDeserializer.java deleted file mode 100644 index 949271a78585..000000000000 --- a/integration/admin-client/src/main/java/org/keycloak/admin/client/deserializer/StreamDeserializer.java +++ /dev/null @@ -1,55 +0,0 @@ -package org.keycloak.admin.client.deserializer; - -import java.io.IOException; -import java.util.List; -import java.util.stream.Stream; - -import com.fasterxml.jackson.core.JsonParser; -import com.fasterxml.jackson.databind.BeanProperty; -import com.fasterxml.jackson.databind.DeserializationContext; -import com.fasterxml.jackson.databind.JavaType; -import com.fasterxml.jackson.databind.JsonDeserializer; -import com.fasterxml.jackson.databind.deser.ContextualDeserializer; - -public class StreamDeserializer extends JsonDeserializer> implements ContextualDeserializer { - - private final JavaType contentType; - - public StreamDeserializer() { - this(null); - } - - public StreamDeserializer(JavaType contentType) { - this.contentType = contentType; - } - - @Override - public Stream deserialize(JsonParser p, DeserializationContext ctxt) throws IOException { - JavaType streamType = ctxt.getContextualType(); - JavaType actualContentType = contentType; - - if (actualContentType == null && streamType != null && streamType.containedTypeCount() == 1) { - actualContentType = streamType.containedType(0); - } - - if (actualContentType == null) { - actualContentType = ctxt.getTypeFactory().constructType(Object.class); - } - - JavaType listType = ctxt.getTypeFactory().constructCollectionType(List.class, actualContentType); - List list = ctxt.readValue(p, listType); - return list.stream(); - } - - @Override - public JsonDeserializer createContextual(DeserializationContext ctxt, BeanProperty property) { - JavaType contextualType = ctxt.getContextualType(); - JavaType actualContentType = null; - - if (contextualType != null && contextualType.containedTypeCount() == 1) { - actualContentType = contextualType.containedType(0); - } - - return new StreamDeserializer(actualContentType); - } -} diff --git a/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/ResteasyClientClassicProvider.java b/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/ResteasyClientClassicProvider.java index 7954dafe5684..a8cb5de4af49 100644 --- a/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/ResteasyClientClassicProvider.java +++ b/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/ResteasyClientClassicProvider.java @@ -20,13 +20,12 @@ import javax.net.ssl.SSLContext; import jakarta.ws.rs.client.Client; -import jakarta.ws.rs.client.ClientBuilder; import jakarta.ws.rs.client.WebTarget; -import org.keycloak.admin.client.ClientBuilderWrapper; import org.keycloak.admin.client.JacksonProvider; -import org.keycloak.admin.client.StreamRxInvokerProvider; + import org.jboss.resteasy.client.jaxrs.ResteasyWebTarget; +import org.jboss.resteasy.client.jaxrs.internal.ResteasyClientBuilderImpl; /** * An implementation of {@link ResteasyClientProvider} based on RESTEasy classic. @@ -35,16 +34,26 @@ public class ResteasyClientClassicProvider implements ResteasyClientProvider { @Override public Client newRestEasyClient(Object customJacksonProvider, SSLContext sslContext, boolean disableTrustManager) { - ClientBuilder clientBuilder = ClientBuilderWrapper.create(sslContext, disableTrustManager); + ResteasyClientBuilderImpl builder = createClientBuilder().sslContext(sslContext); if (customJacksonProvider != null) { - clientBuilder.register(customJacksonProvider, 100); + builder.register(customJacksonProvider, 100); } else { - clientBuilder.register(JacksonProvider.class, 100); + builder.register(JacksonProvider.class, 100); + } + + if (disableTrustManager) { + builder.disableTrustManager(); } - clientBuilder.register(StreamRxInvokerProvider.class); + return builder.build(); + } - return clientBuilder.build(); + /** + * Create a new client builder instance with common defaults and provider registrations + * @return a new client builder + */ + public static ResteasyClientBuilderImpl createClientBuilder() { + return new ResteasyClientBuilderImpl().connectionPoolSize(10).register(StreamMessageBodyReader.class); } @Override diff --git a/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java b/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java new file mode 100644 index 000000000000..f1ae97aaf5b5 --- /dev/null +++ b/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java @@ -0,0 +1,81 @@ +package org.keycloak.admin.client.spi; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.lang.annotation.Annotation; +import java.lang.reflect.ParameterizedType; +import java.lang.reflect.Proxy; +import java.lang.reflect.Type; +import java.nio.charset.StandardCharsets; +import java.util.Iterator; +import java.util.Spliterator; +import java.util.Spliterators; +import java.util.stream.Stream; +import java.util.stream.StreamSupport; + +import jakarta.ws.rs.WebApplicationException; +import jakarta.ws.rs.core.MediaType; +import jakarta.ws.rs.core.MultivaluedMap; +import jakarta.ws.rs.ext.MessageBodyReader; + +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import com.fasterxml.jackson.core.ObjectCodec; +import org.jboss.resteasy.plugins.providers.sse.EventInput; +import org.jboss.resteasy.specimpl.ResponseBuilderImpl; + +/** + * Provides a way to read Streams incrementally - valid only for Resteasy Classic + */ +public class StreamMessageBodyReader implements MessageBodyReader> { + + @Override + public boolean isReadable(Class type, Type genericType, Annotation[] annotations, MediaType mediaType) { + return type == Stream.class; + } + + @Override + public Stream readFrom(Class> type, Type genericType, Annotation[] annotations, MediaType mediaType, + MultivaluedMap httpHeaders, InputStream entityStream) + throws IOException, WebApplicationException { + + // workaround to obtain an ObjectCodec - it would be better to work directly from a mapper, but + // there's no good assumption to make about how to obtain that + ObjectCodec codec = new ResponseBuilderImpl() + .entity(new ByteArrayInputStream("[]".getBytes(StandardCharsets.UTF_8))).status(200).type(mediaType) + .build().readEntity(JsonParser.class).getCodec(); + + JsonParser parser = codec.getFactory().createParser(entityStream); + JsonToken token = parser.nextToken(); + if (token == null) { + return null; + } + if (token != JsonToken.START_ARRAY) { + entityStream.close(); + throw new IOException("Expected Array"); + } + token = parser.nextToken(); + if (token == JsonToken.END_ARRAY) { + return Stream.of(); + } + if (token != JsonToken.START_OBJECT) { + entityStream.close(); + throw new IOException("Expected Object"); + } + + Iterator iter = codec.readValues(parser, (Class)((ParameterizedType)genericType).getActualTypeArguments()[0]); + + Stream targetStream = StreamSupport.stream( + Spliterators.spliteratorUnknownSize(iter, Spliterator.ORDERED), + false); + // return a Stream also marked as an EventInput to prevent the premature closure of the result + return (Stream) Proxy.newProxyInstance(EventInput.class.getClassLoader(), new Class[] {Stream.class, EventInput.class}, (proxy, method, args) -> { + if (method.getName().equals("close")) { + entityStream.close(); + } + return method.invoke(targetStream, args); + }); + } + +} From a7691c79a06a8ed48bbbc7d641f5ca1f9e0a03b5 Mon Sep 17 00:00:00 2001 From: Steven Hawkins Date: Fri, 3 Apr 2026 12:40:48 -0400 Subject: [PATCH 3/5] Apply suggestion from @Copilot Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Signed-off-by: Steven Hawkins --- .../admin/client/spi/StreamMessageBodyReader.java | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java b/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java index f1ae97aaf5b5..69b885038ff5 100644 --- a/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java +++ b/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java @@ -64,8 +64,15 @@ public Stream readFrom(Class> type, Type genericType, Annotation[] throw new IOException("Expected Object"); } - Iterator iter = codec.readValues(parser, (Class)((ParameterizedType)genericType).getActualTypeArguments()[0]); + Class elementType = Object.class; + if (genericType instanceof ParameterizedType) { + Type[] typeArguments = ((ParameterizedType) genericType).getActualTypeArguments(); + if (typeArguments.length > 0 && typeArguments[0] instanceof Class) { + elementType = (Class) typeArguments[0]; + } + } + Iterator iter = codec.readValues(parser, elementType); Stream targetStream = StreamSupport.stream( Spliterators.spliteratorUnknownSize(iter, Spliterator.ORDERED), false); From 59db65afaeb71f221dfef518acefd8cb5420700d Mon Sep 17 00:00:00 2001 From: Steve Hawkins Date: Thu, 9 Apr 2026 19:10:26 -0400 Subject: [PATCH 4/5] using onClose instead of the proxy close Signed-off-by: Steve Hawkins --- .../client/spi/StreamMessageBodyReader.java | 22 ++++++++++++++----- 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java b/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java index 69b885038ff5..c246b8e1326d 100644 --- a/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java +++ b/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java @@ -3,10 +3,12 @@ import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; +import java.io.UncheckedIOException; import java.lang.annotation.Annotation; import java.lang.reflect.ParameterizedType; import java.lang.reflect.Proxy; import java.lang.reflect.Type; +import java.lang.reflect.WildcardType; import java.nio.charset.StandardCharsets; import java.util.Iterator; import java.util.Spliterator; @@ -67,20 +69,28 @@ public Stream readFrom(Class> type, Type genericType, Annotation[] Class elementType = Object.class; if (genericType instanceof ParameterizedType) { Type[] typeArguments = ((ParameterizedType) genericType).getActualTypeArguments(); - if (typeArguments.length > 0 && typeArguments[0] instanceof Class) { - elementType = (Class) typeArguments[0]; + if (typeArguments.length > 0) { + if (typeArguments[0] instanceof WildcardType) { + typeArguments = ((WildcardType)typeArguments[0]).getUpperBounds(); + } + if (typeArguments.length > 0 && typeArguments[0] instanceof Class) { + elementType = (Class) typeArguments[0]; + } } } Iterator iter = codec.readValues(parser, elementType); Stream targetStream = StreamSupport.stream( Spliterators.spliteratorUnknownSize(iter, Spliterator.ORDERED), - false); - // return a Stream also marked as an EventInput to prevent the premature closure of the result - return (Stream) Proxy.newProxyInstance(EventInput.class.getClassLoader(), new Class[] {Stream.class, EventInput.class}, (proxy, method, args) -> { - if (method.getName().equals("close")) { + false).onClose(() -> { + try { entityStream.close(); + } catch (IOException e) { + throw new UncheckedIOException(e); } + }); + // return a Stream also marked as an EventInput to prevent the premature closure of the result + return (Stream) Proxy.newProxyInstance(EventInput.class.getClassLoader(), new Class[] {Stream.class, EventInput.class}, (proxy, method, args) -> { return method.invoke(targetStream, args); }); } From 5412c56848492ff1406e0799fd7a89ac0b63e4bf Mon Sep 17 00:00:00 2001 From: Steven Hawkins Date: Tue, 26 May 2026 09:19:26 -0400 Subject: [PATCH 5/5] Update integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java Co-authored-by: Peter Zaoral Signed-off-by: Steven Hawkins --- .../keycloak/admin/client/spi/StreamMessageBodyReader.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java b/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java index c246b8e1326d..21258a7ce140 100644 --- a/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java +++ b/integration/admin-client/src/main/java/org/keycloak/admin/client/spi/StreamMessageBodyReader.java @@ -5,6 +5,7 @@ import java.io.InputStream; import java.io.UncheckedIOException; import java.lang.annotation.Annotation; +import java.lang.reflect.InvocationTargetException; import java.lang.reflect.ParameterizedType; import java.lang.reflect.Proxy; import java.lang.reflect.Type; @@ -91,7 +92,11 @@ public Stream readFrom(Class> type, Type genericType, Annotation[] }); // return a Stream also marked as an EventInput to prevent the premature closure of the result return (Stream) Proxy.newProxyInstance(EventInput.class.getClassLoader(), new Class[] {Stream.class, EventInput.class}, (proxy, method, args) -> { - return method.invoke(targetStream, args); + try { + return method.invoke(targetStream, args); + } catch (InvocationTargetException e) { + throw e.getCause(); + } }); }