Modified calls to Content.Source.demand() to specify InvocationType t…

…o be NON_BLOCKING. Signed-off-by: Simone Bordet <[email protected]>
jetty-project · Nov 1, 2024 · 6dc3eda · 6dc3eda
1 parent f8c158d
commit 6dc3eda
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 3 deletions.
diff --git a/src/main/java/org/eclipse/jetty/reactive/client/internal/ResponseEventPublisher.java b/src/main/java/org/eclipse/jetty/reactive/client/internal/ResponseEventPublisher.java
@@ -16,13 +16,13 @@
 package org.eclipse.jetty.reactive.client.internal;
 
 import java.nio.ByteBuffer;
-
 import org.eclipse.jetty.client.Response;
 import org.eclipse.jetty.client.Result;
 import org.eclipse.jetty.http.HttpField;
 import org.eclipse.jetty.io.Content;
 import org.eclipse.jetty.reactive.client.ReactiveRequest;
 import org.eclipse.jetty.reactive.client.ReactiveResponse;
+import org.eclipse.jetty.util.thread.Invocable;
 
 public class ResponseEventPublisher extends AbstractEventPublisher<ReactiveResponse.Event> implements Response.Listener {
     private final ReactiveRequest request;
@@ -56,7 +56,7 @@ public void onContent(Response response, Content.Chunk chunk, Runnable demander)
 
     @Override
     public void onContentSource(Response response, Content.Source source) {
-        Runnable reader = new Runnable() {
+        Runnable reader = new Invocable.Task.Abstract(Invocable.InvocationType.NON_BLOCKING) {
             @Override
             public void run() {
                 while (true) {

diff --git a/src/main/java/org/eclipse/jetty/reactive/client/internal/ResponseListenerProcessor.java b/src/main/java/org/eclipse/jetty/reactive/client/internal/ResponseListenerProcessor.java
@@ -23,6 +23,7 @@
 import org.eclipse.jetty.reactive.client.ReactiveRequest;
 import org.eclipse.jetty.reactive.client.ReactiveResponse;
 import org.eclipse.jetty.util.thread.AutoLock;
+import org.eclipse.jetty.util.thread.Invocable;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.slf4j.Logger;
@@ -169,7 +170,7 @@ public String toString() {
      * <p>Publishes response {@link Content.Chunk}s to the application
      * {@code BiFunction} given to {@link ReactiveRequest#response(BiFunction)}.</p>
      */
-    private static class ContentPublisher extends QueuedSinglePublisher<Content.Chunk> implements Runnable {
+    private static class ContentPublisher extends QueuedSinglePublisher<Content.Chunk> implements Invocable.Task {
         private volatile Content.Source contentSource;
 
         private void accept(Content.Source source) {
@@ -195,6 +196,11 @@ public void run() {
             }
         }
 
+        @Override
+        public InvocationType getInvocationType() {
+            return InvocationType.NON_BLOCKING;
+        }
+
         private void read(Content.Source source) {
             while (true) {
                 if (!hasDemand()) {