Revert "Merge pull request #19664 from vespa-engine/jonmv/streaming-doc-v1-visit"

This reverts commit b5d4b42caabf2c41fcbd8b21814819aae77dc7aa, reversing
changes made to 9abe019606f2367b05e4e13d796de65dddf7c449.

3 files changed, 83 insertions, 92 deletions

diff --git a/container-core/src/main/java/com/yahoo/container/jdisc/MaxPendingContentChannelOutputStream.java b/container-core/src/main/java/com/yahoo/container/jdisc/MaxPendingContentChannelOutputStream.java
deleted file mode 100644
index aec4eeecd7b..00000000000
--- a/container-core/src/main/java/com/yahoo/container/jdisc/MaxPendingContentChannelOutputStream.java
+++ /dev/null
@@ -1,92 +0,0 @@
-// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
-package com.yahoo.container.jdisc;
-
-import com.yahoo.jdisc.handler.CompletionHandler;
-import com.yahoo.jdisc.handler.ContentChannel;
-
-import java.io.IOException;
-import java.io.InterruptedIOException;
-import java.nio.ByteBuffer;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
-
-/**
- * @author baldersheim
- */
-public class MaxPendingContentChannelOutputStream extends ContentChannelOutputStream {
-
-    private final long maxPending;
-    private final AtomicLong sent = new AtomicLong(0);
-    private final AtomicLong acked = new AtomicLong(0);
-
-    public MaxPendingContentChannelOutputStream(ContentChannel endpoint, long maxPending) {
-        super(endpoint);
-        this.maxPending = maxPending;
-    }
-
-    private long pendingBytes() {
-        return sent.get() - acked.get();
-    }
-
-    private class TrackCompletion implements CompletionHandler {
-
-        private final long written;
-        private final AtomicBoolean replied = new AtomicBoolean(false);
-
-        TrackCompletion(long written) {
-            this.written = written;
-            sent.addAndGet(written);
-        }
-
-        @Override
-        public void completed() {
-            if (!replied.getAndSet(true)) {
-                acked.addAndGet(written);
-            }
-        }
-
-        @Override
-        public void failed(Throwable t) {
-            if (!replied.getAndSet(true)) {
-                acked.addAndGet(written);
-            }
-        }
-
-    }
-
-    @Override
-    public void send(ByteBuffer src) throws IOException {
-        try {
-            stallWhilePendingAbove(maxPending);
-        }
-        catch (InterruptedException ignored) {
-            throw new InterruptedIOException("Interrupted waiting for IO");
-        }
-        CompletionHandler pendingTracker = new TrackCompletion(src.remaining());
-        try {
-            send(src, pendingTracker);
-        }
-        catch (Throwable throwable) {
-            pendingTracker.failed(throwable);
-            throw throwable;
-        }
-    }
-
-    private void stallWhilePendingAbove(long pending) throws InterruptedException {
-        while (pendingBytes() > pending) {
-            Thread.sleep(1);
-        }
-    }
-
-    @Override
-    public void flush() throws IOException {
-        super.flush();
-        try {
-            stallWhilePendingAbove(0);
-        }
-        catch (InterruptedException e) {
-            throw new InterruptedIOException("Interrupted waiting for IO");
-        }
-    }
-
-}
diff --git a/container-core/src/main/java/com/yahoo/container/jdisc/ThreadedHttpRequestHandler.java b/container-core/src/main/java/com/yahoo/container/jdisc/ThreadedHttpRequestHandler.java
index 0c3c1e2120b..0bfe4afe07d 100644
--- a/container-core/src/main/java/com/yahoo/container/jdisc/ThreadedHttpRequestHandler.java
+++ b/container-core/src/main/java/com/yahoo/container/jdisc/ThreadedHttpRequestHandler.java
@@ -10,6 +10,9 @@ import com.yahoo.jdisc.handler.ContentChannel;
 import com.yahoo.jdisc.handler.UnsafeContentInputStream;
 import com.yahoo.jdisc.handler.ResponseHandler;
 
+import java.io.InterruptedIOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.logging.Level;
 
 import java.io.IOException;
@@ -250,4 +253,81 @@ public abstract class ThreadedHttpRequestHandler extends ThreadedRequestHandler
     }
 
 
+    /**
+     * @author baldersheim
+     */
+    static class MaxPendingContentChannelOutputStream extends ContentChannelOutputStream {
+        private final long maxPending;
+        private final AtomicLong sent = new AtomicLong(0);
+        private final AtomicLong acked = new AtomicLong(0);
+
+        public MaxPendingContentChannelOutputStream(ContentChannel endpoint, long maxPending) {
+            super(endpoint);
+            this.maxPending = maxPending;
+        }
+
+        private long pendingBytes() {
+            return sent.get() - acked.get();
+        }
+
+        private class TrackCompletion implements CompletionHandler {
+
+            private final long written;
+            private final AtomicBoolean replied = new AtomicBoolean(false);
+
+            TrackCompletion(long written) {
+                this.written = written;
+                sent.addAndGet(written);
+            }
+
+            @Override
+            public void completed() {
+                if ( ! replied.getAndSet(true)) {
+                    acked.addAndGet(written);
+                }
+            }
+
+            @Override
+            public void failed(Throwable t) {
+                if ( ! replied.getAndSet(true)) {
+                    acked.addAndGet(written);
+                }
+            }
+        }
+
+        @Override
+        public void send(ByteBuffer src) throws IOException {
+            try {
+                stallWhilePendingAbove(maxPending);
+            } catch (InterruptedException ignored) {
+                throw new InterruptedIOException("Interrupted waiting for IO");
+            }
+            CompletionHandler pendingTracker = new TrackCompletion(src.remaining());
+            try {
+                send(src, pendingTracker);
+            } catch (Throwable throwable) {
+                pendingTracker.failed(throwable);
+                throw throwable;
+            }
+        }
+
+        private void stallWhilePendingAbove(long pending) throws InterruptedException {
+            while (pendingBytes() > pending) {
+                Thread.sleep(1);
+            }
+        }
+
+        @Override
+        public void flush() throws IOException {
+            super.flush();
+            try {
+                stallWhilePendingAbove(0);
+            }
+            catch (InterruptedException e) {
+                throw new InterruptedIOException("Interrupted waiting for IO");
+            }
+        }
+
+    }
+
 }
diff --git a/container-core/src/test/java/com/yahoo/container/jdisc/ThreadedRequestHandlerTestCase.java b/container-core/src/test/java/com/yahoo/container/jdisc/ThreadedRequestHandlerTestCase.java
index a9b16799aea..d1036ce0e45 100644
--- a/container-core/src/test/java/com/yahoo/container/jdisc/ThreadedRequestHandlerTestCase.java
+++ b/container-core/src/test/java/com/yahoo/container/jdisc/ThreadedRequestHandlerTestCase.java
@@ -1,6 +1,7 @@
 // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.container.jdisc;
 
+import com.yahoo.container.jdisc.ThreadedHttpRequestHandler.MaxPendingContentChannelOutputStream;
 import com.yahoo.jdisc.Request;
 import com.yahoo.jdisc.Response;
 import com.yahoo.jdisc.application.ContainerBuilder;
@@ -19,9 +20,11 @@ import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.Phaser;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.junit.Assert.*;