package org.apache.camel.component.undertow;

import org.junit.Test;
import org.xnio.XnioIoThread;
import org.xnio.channels.EmptyStreamSourceChannel;
import org.xnio.channels.StreamSourceChannel;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertThat;

public class DefaultUndertowHttpBindingTest {

    @Test(timeout = 1000)
    public void readEntireDelayedPayload() throws Exception {
        String[] delayedPayloads = new String[] {
                "chunk",
        };

        StreamSourceChannel source = source(delayedPayloads);

        DefaultUndertowHttpBinding binding = new DefaultUndertowHttpBinding();
        String result = new String(binding.readFromChannel(source));

        assertThat(result, is(delayedPayloads[0]));
    }

    @Test(timeout = 1000)
    public void readEntireMultiDelayedPayload() throws Exception {
        String[] delayedPayloads = new String[] {
                "first ",
                "second",
        };

        StreamSourceChannel source = source(delayedPayloads);

        DefaultUndertowHttpBinding binding = new DefaultUndertowHttpBinding();
        String result = new String(binding.readFromChannel(source));

        assertThat(result, is(
                Stream.of(delayedPayloads)
                        .collect(Collectors.joining())));
    }

    @Test(timeout = 1000)
    public void readEntireMultiDelayedWithPausePayload() throws Exception {
        String[] delayedPayloads = new String[] {
                "first ",
                "",
                "second",
        };

        StreamSourceChannel source = source(delayedPayloads);

        DefaultUndertowHttpBinding binding = new DefaultUndertowHttpBinding();
        String result = new String(binding.readFromChannel(source));

        assertThat(result, is(
                Stream.of(delayedPayloads)
                        .collect(Collectors.joining())));
    }

    private StreamSourceChannel source(final String[] delayedPayloads) {
        Thread sourceThread = Thread.currentThread();

        return new EmptyStreamSourceChannel(thread()) {
            int chunk = 0;

            @Override
            public int read(ByteBuffer dst) throws IOException {
                // can only read payloads in the reader thread
                if (sourceThread != Thread.currentThread()) {
                    if (chunk < delayedPayloads.length) {
                        byte[] delayedPayload = delayedPayloads[chunk].getBytes();
                        dst.put(delayedPayload);
                        chunk++;
                        return delayedPayload.length;
                    }
                    return -1;
                }
                return 0;
            }

            @Override
            public void resumeReads() {
                /**
                 * {@link io.undertow.server.HttpServerExchange.ReadDispatchChannel} delays resumes in the main thread
                 */
                if (sourceThread != Thread.currentThread()) {
                    super.resumeReads();
                }
            }
        };
    }

    private XnioIoThread thread() {
        ExecutorService executor = Executors.newFixedThreadPool(1);
        return new XnioIoThread(null, 0) {
            @Override
            public void execute(Runnable runnable) {
                executor.execute(runnable);
            }

            @Override
            public Key executeAfter(Runnable runnable, long l, TimeUnit timeUnit) {
                execute(runnable);
                return null;
            }

            @Override
            public Key executeAtInterval(Runnable runnable, long l, TimeUnit timeUnit) {
                execute(runnable);
                return null;
            }
        };
    }
}
@@ -518,16 +518,53 @@
 ertThat;
+%0Aimport static org.junit.Assert.fail;
 %0A%0Apublic
@@ -734,32 +734,52 @@
  new String%5B%5D %7B%0A
+                %22%22,%0A
                 
@@ -1015,33 +1015,34 @@
 );%0A%0A        
-assertTha
+checkResul
 t(result, is
@@ -1039,19 +1039,16 @@
 result, 
-is(
 delayedP
@@ -1058,12 +1058,8 @@
 oads
-%5B0%5D)
 );%0A 
@@ -1199,32 +1199,52 @@
  new String%5B%5D %7B%0A
+                %22%22,%0A
                 
@@ -1495,32 +1495,157 @@
 nnel(source));%0A%0A
+        checkResult(result, delayedPayloads);%0A    %7D%0A%0A    private void checkResult(String result, String%5B%5D delayedPayloads) %7B%0A
         assertTh
@@ -1911,32 +1911,52 @@
  new String%5B%5D %7B%0A
+                %22%22,%0A
                 
@@ -2247,129 +2247,43 @@
     
-assertThat(result, is(%0A                Stream.of(delayedPayloads)%0A                        .collect(Collectors.joining()))
+checkResult(result, delayedPayloads
 );%0A 
@@ -2498,16 +2498,111 @@
 unk = 0;
+%0A            boolean mustWait = false;  // make sure that the caller is not spinning on read==0
 %0A%0A      
@@ -2702,121 +2702,96 @@
     
-// can only read payloads in the reader thread%0A                if (sourceThread != Thread.currentThread()) %7B%0A    
+if (mustWait) %7B%0A                    fail(%22must wait before reading%22);%0A                %7D%0A
     
@@ -2860,20 +2860,16 @@
         
-  
-  
 byte%5B%5D d
@@ -2939,20 +2939,16 @@
         
-    
 dst.put(
@@ -2964,20 +2964,16 @@
 yload);%0A
-    
         
@@ -3013,27 +3013,20 @@
         
-    return 
+if (
 delayedP
@@ -3038,16 +3038,64 @@
 d.length
+ == 0) %7B%0A                        mustWait = true
 ;%0A      
@@ -3141,10 +3141,29 @@
 urn 
--1
+delayedPayload.length
 ;%0A  
@@ -3201,17 +3201,18 @@
  return 
-0
+-1
 ;%0A      
@@ -3567,32 +3567,203 @@
  %7D%0A            %7D
+%0A%0A            @Override%0A            public void awaitReadable() throws IOException %7B%0A                mustWait = false;%0A                super.awaitReadable();%0A            %7D
 %0A        %7D;%0A    

correct tests and ensure that reader's wait for data