Use default content type when X-SSE header not set

quarkusio · Oct 12, 2023 · a6d9c09 · a6d9c09
1 parent 20075d7
commit a6d9c09
Show file tree

Hide file tree

Showing 4 changed files with 52 additions and 7 deletions.
diff --git a/...n/deployment/src/test/java/io/quarkus/rest/client/reactive/jackson/test/MultiSseTest.java b/...n/deployment/src/test/java/io/quarkus/rest/client/reactive/jackson/test/MultiSseTest.java
@@ -112,6 +112,20 @@ void shouldRestStreamElementTypeOverwriteProducesAtClassLevel() {
                                 .containsExactly(new Dto("foo", "bar"), new Dto("chocolate", "bar")));
     }
 
+    @Test
+    void shouldUseDefaultContentTypeWhenXSseHeaderNotPresent() {
+        var resultList = new CopyOnWriteArrayList<>();
+        QuarkusRestClientBuilder.newBuilder().baseUri(uri)
+                .build(SeeWithRestStreamElementTypeClientWithoutSpecialHeader.class)
+                .getJson()
+                .subscribe()
+                .with(resultList::add);
+        await().atMost(5, TimeUnit.SECONDS)
+                .untilAsserted(
+                        () -> assertThat(resultList)
+                                .containsExactly(new Dto("foo", "bar")));
+    }
+
     private SseClient createClient() {
         return QuarkusRestClientBuilder.newBuilder()
                 .baseUri(uri)
@@ -200,6 +214,26 @@ public interface SeeWithRestStreamElementTypeClient {
         Multi<Dto> getJson();
     }
 
+    @Path("/sse-rest-stream-element-type-no-x-sse")
+    public static class SseWithRestStreamElementTypeResourceWithoutSpecialHeader {
+        @GET
+        @Produces(MediaType.SERVER_SENT_EVENTS)
+        @Path("/json")
+        public Multi<String> getJson() {
+            return Multi.createFrom().items("{\"name\": \"foo\", \"value\": \"bar\"}");
+        }
+    }
+
+    @RegisterRestClient
+    @Path("/sse-rest-stream-element-type-no-x-sse")
+    @Produces(MediaType.APPLICATION_JSON)
+    public interface SeeWithRestStreamElementTypeClientWithoutSpecialHeader {
+        @GET
+        @Path("/json")
+        @RestStreamElementType(MediaType.APPLICATION_JSON)
+        Multi<Dto> getJson();
+    }
+
     public static class Dto {
         public String name;
         public String value;

diff --git a/...ve/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/MultiInvoker.java b/...ve/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/MultiInvoker.java
@@ -125,7 +125,9 @@ public <R> Multi<R> method(String name, Entity<?> entity, GenericType<R> respons
                     if (!emitter.isCancelled()) {
                         if (response.getStatus() == 200
                                 && MediaType.SERVER_SENT_EVENTS_TYPE.isCompatible(response.getMediaType())) {
-                            registerForSse(multiRequest, responseType, response, vertxResponse);
+                            registerForSse(multiRequest, responseType, response, vertxResponse,
+                                    (String) restClientRequestContext.getProperties()
+                                            .get(RestClientRequestContext.DEFAULT_CONTENT_TYPE_PROP));
                         } else if (response.getStatus() == 200
                                 && RestMediaType.APPLICATION_STREAM_JSON_TYPE.isCompatible(response.getMediaType())) {
                             registerForJsonStream(multiRequest, restClientRequestContext, responseType, response,
@@ -152,12 +154,12 @@ private boolean isNewlineDelimited(ResponseImpl response) {
     private <R> void registerForSse(MultiRequest<? super R> multiRequest,
             GenericType<R> responseType,
             Response response,
-            HttpClientResponse vertxResponse) {
+            HttpClientResponse vertxResponse, String defaultContentType) {
         // honestly, isn't reconnect contradictory with completion?
         // FIXME: Reconnect settings?
         // For now we don't want multi to reconnect
         SseEventSourceImpl sseSource = new SseEventSourceImpl(invocationBuilder.getTarget(),
-                invocationBuilder, Integer.MAX_VALUE, TimeUnit.SECONDS);
+                invocationBuilder, Integer.MAX_VALUE, TimeUnit.SECONDS, defaultContentType);
 
         multiRequest.onCancel(sseSource::close);
         sseSource.register(event -> {

diff --git a/...ent/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/SseEventSourceImpl.java b/...ent/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/SseEventSourceImpl.java
@@ -42,14 +42,19 @@ public class SseEventSourceImpl implements SseEventSource, Handler<Long> {
 
     public SseEventSourceImpl(WebTargetImpl webTarget, Invocation.Builder invocationBuilder,
             long reconnectDelay, TimeUnit reconnectUnit) {
+        this(webTarget, invocationBuilder, reconnectDelay, reconnectUnit, null);
+    }
+
+    public SseEventSourceImpl(WebTargetImpl webTarget, Invocation.Builder invocationBuilder,
+            long reconnectDelay, TimeUnit reconnectUnit, String defaultContentType) {
         // tests set a null endpoint
         Objects.requireNonNull(reconnectUnit);
         if (reconnectDelay <= 0)
             throw new IllegalArgumentException("Delay must be > 0: " + reconnectDelay);
         this.webTarget = webTarget;
         this.reconnectDelay = reconnectDelay;
         this.reconnectUnit = reconnectUnit;
-        this.sseParser = new SseParser(this);
+        this.sseParser = new SseParser(this, defaultContentType);
         this.invocationBuilder = invocationBuilder;
     }
 
@@ -136,7 +141,9 @@ private void registerOnClient(HttpClientResponse vertxClientResponse) {
         vertxClientResponse.request().exceptionHandler(null);
         connection = vertxClientResponse.request().connection();
         String sseContentTypeHeader = vertxClientResponse.getHeader(CommonSseUtil.SSE_CONTENT_TYPE);
-        sseParser.setSseContentTypeHeader(sseContentTypeHeader);
+        if ((sseContentTypeHeader != null) && !sseContentTypeHeader.isEmpty()) {
+            sseParser.setSseContentTypeHeader(sseContentTypeHeader);
+        }
         // we don't add a closeHandler handler on the connection as it can race with this handler
         // and close before the emitter emits anything
         // see: https://github.com/quarkusio/quarkus/pull/16438

diff --git a/...ctive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/SseParser.java b/...ctive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/SseParser.java
@@ -50,7 +50,8 @@ public class SseParser implements Handler<Buffer> {
      */
     private String contentType;
     /**
-     * The content type we're reading. Defaults to the X-Sse-Element-Type header
+     * The content type we're reading. If the X-Sse-Element-Type header is not set, then it defaults to the declared @Produces
+     * (if any)
      */
     private String contentTypeHeader;
     /**
@@ -67,8 +68,9 @@ public class SseParser implements Handler<Buffer> {
     private long eventReconnectTime = SseEvent.RECONNECT_NOT_SET;
     private SseEventSourceImpl sseEventSource;
 
-    public SseParser(SseEventSourceImpl sseEventSource) {
+    public SseParser(SseEventSourceImpl sseEventSource, String defaultContentType) {
         this.sseEventSource = sseEventSource;
+        this.contentTypeHeader = defaultContentType;
     }
 
     public void setSseContentTypeHeader(String sseContentTypeHeader) {