jetty-9 added IOFuture test
This commit is contained in:
parent
8f8c3f7760
commit
6a5fdefa07
|
@ -62,15 +62,17 @@ import java.util.concurrent.Future;
|
||||||
* <h3>Hybrid write</h3>
|
* <h3>Hybrid write</h3>
|
||||||
* <pre>
|
* <pre>
|
||||||
* IOFuture write = endpoint.write(buffer);
|
* IOFuture write = endpoint.write(buffer);
|
||||||
* if (write.isReady())
|
* // wait a little bit
|
||||||
* // write next
|
* if (!write.block(10,TimeUnit.MILLISECONDS))
|
||||||
* else
|
* {
|
||||||
|
* // still not ready, so organize a callback
|
||||||
* write.setHandler(new IOCallback()
|
* write.setHandler(new IOCallback()
|
||||||
* {
|
* {
|
||||||
* public void onReady() { ... }
|
* public void onReady() { ... }
|
||||||
* public void onFail(IOException e) { ... }
|
* public void onFail(IOException e) { ... }
|
||||||
* public void onTimeout() { ... }
|
* public void onTimeout() { ... }
|
||||||
* }
|
* });
|
||||||
|
* ...
|
||||||
* </pre>
|
* </pre>
|
||||||
*
|
*
|
||||||
* <h2>Compatibility Notes</h2>
|
* <h2>Compatibility Notes</h2>
|
||||||
|
|
|
@ -1,6 +1,7 @@
|
||||||
package org.eclipse.jetty.io;
|
package org.eclipse.jetty.io;
|
||||||
|
|
||||||
import java.util.concurrent.ExecutionException;
|
import java.util.concurrent.ExecutionException;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
|
||||||
import org.eclipse.jetty.util.thread.ThreadPool;
|
import org.eclipse.jetty.util.thread.ThreadPool;
|
||||||
|
|
||||||
|
@ -44,6 +45,12 @@ public class CompleteIOFuture implements IOFuture
|
||||||
isReady();
|
isReady();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public boolean block(long timeout, TimeUnit units) throws ExecutionException
|
||||||
|
{
|
||||||
|
return isReady();
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void setCallback(final Callback callback)
|
public void setCallback(final Callback callback)
|
||||||
{
|
{
|
||||||
|
|
|
@ -1,6 +1,8 @@
|
||||||
package org.eclipse.jetty.io;
|
package org.eclipse.jetty.io;
|
||||||
|
|
||||||
import java.util.concurrent.ExecutionException;
|
import java.util.concurrent.ExecutionException;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
import java.util.concurrent.TimeoutException;
|
||||||
|
|
||||||
/* ------------------------------------------------------------ */
|
/* ------------------------------------------------------------ */
|
||||||
/* ------------------------------------------------------------ */
|
/* ------------------------------------------------------------ */
|
||||||
|
@ -9,14 +11,17 @@ import java.util.concurrent.ExecutionException;
|
||||||
* <p>
|
* <p>
|
||||||
* This interface make the future status of an IO operation available via
|
* This interface make the future status of an IO operation available via
|
||||||
* polling ({@link #isReady()}, blocking ({@link #block()} or callback ({@link #setCallback(Callback)}
|
* polling ({@link #isReady()}, blocking ({@link #block()} or callback ({@link #setCallback(Callback)}
|
||||||
*
|
* <p>
|
||||||
|
* This interface does not directly support a timeout for blocking. If an IO operation times out, then
|
||||||
|
* this will be indicated via this interface by an {@link ExecutionException} containing a
|
||||||
|
* {@link TimeoutException} (or similar).
|
||||||
*/
|
*/
|
||||||
public interface IOFuture
|
public interface IOFuture
|
||||||
{
|
{
|
||||||
/* ------------------------------------------------------------ */
|
/* ------------------------------------------------------------ */
|
||||||
/** Indicate if this Future is complete.
|
/** Indicate if this Future is complete.
|
||||||
* If this future has completed by becoming ready, excepting or timeout.
|
* If this future has completed by becoming ready, excepting or timeout.
|
||||||
* @return True if this future has completed by becoming ready, excepting or timeout.
|
* @return True if this future has completed by becoming ready or excepting.
|
||||||
*/
|
*/
|
||||||
boolean isComplete();
|
boolean isComplete();
|
||||||
|
|
||||||
|
@ -39,12 +44,23 @@ public interface IOFuture
|
||||||
/* ------------------------------------------------------------ */
|
/* ------------------------------------------------------------ */
|
||||||
/** Block until complete.
|
/** Block until complete.
|
||||||
* <p>This call blocks the calling thread until this AsyncIO is ready or
|
* <p>This call blocks the calling thread until this AsyncIO is ready or
|
||||||
* an exception or until a timeout due to {@link EndPoint#getMaxIdleTime()}.
|
* an exception.
|
||||||
* @throws InterruptedException if interrupted while blocking
|
* @throws InterruptedException if interrupted while blocking
|
||||||
* @throws ExecutionException If any exception occurs during the IO operation
|
* @throws ExecutionException If any exception occurs during the IO operation
|
||||||
*/
|
*/
|
||||||
void block() throws InterruptedException, ExecutionException;
|
void block() throws InterruptedException, ExecutionException;
|
||||||
|
|
||||||
|
/* ------------------------------------------------------------ */
|
||||||
|
/** Block until complete or timeout
|
||||||
|
* <p>This call blocks the calling thread until this AsyncIO is ready or
|
||||||
|
* an exception or a timeout. In the case of the timeout, the IO operation is not affected
|
||||||
|
* and can still continue to completion and this IOFuture is still usable.
|
||||||
|
* @return true if the IOFuture completed or false if it timedout.
|
||||||
|
* @throws InterruptedException if interrupted while blocking
|
||||||
|
* @throws ExecutionException If any exception occurs during the IO operation
|
||||||
|
*/
|
||||||
|
boolean block(long timeout, TimeUnit units) throws InterruptedException, ExecutionException;
|
||||||
|
|
||||||
/* ------------------------------------------------------------ */
|
/* ------------------------------------------------------------ */
|
||||||
/** Set an IOCallback.
|
/** Set an IOCallback.
|
||||||
* Set an {@link Callback} instance to be called when the IO operation is ready or if
|
* Set an {@link Callback} instance to be called when the IO operation is ready or if
|
||||||
|
|
|
@ -1,6 +1,7 @@
|
||||||
package org.eclipse.jetty.io;
|
package org.eclipse.jetty.io;
|
||||||
|
|
||||||
import java.util.concurrent.ExecutionException;
|
import java.util.concurrent.ExecutionException;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
import java.util.concurrent.locks.Condition;
|
import java.util.concurrent.locks.Condition;
|
||||||
import java.util.concurrent.locks.Lock;
|
import java.util.concurrent.locks.Lock;
|
||||||
import java.util.concurrent.locks.ReentrantLock;
|
import java.util.concurrent.locks.ReentrantLock;
|
||||||
|
@ -175,6 +176,23 @@ public class RecycledIOFuture implements IOFuture
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public boolean block(long timeout, TimeUnit units) throws InterruptedException, ExecutionException
|
||||||
|
{
|
||||||
|
_lock.lock();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
if (!_complete)
|
||||||
|
_block.await(timeout,units);
|
||||||
|
return isReady();
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_lock.unlock();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void setCallback(Callback callback)
|
public void setCallback(Callback callback)
|
||||||
{
|
{
|
||||||
|
|
|
@ -0,0 +1,355 @@
|
||||||
|
package org.eclipse.jetty.io;
|
||||||
|
|
||||||
|
import static org.hamcrest.number.OrderingComparison.*;
|
||||||
|
import static org.junit.Assert.assertEquals;
|
||||||
|
import static org.junit.Assert.assertFalse;
|
||||||
|
import static org.junit.Assert.assertThat;
|
||||||
|
import static org.junit.Assert.assertTrue;
|
||||||
|
|
||||||
|
import java.util.concurrent.ExecutionException;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
import java.util.concurrent.atomic.AtomicBoolean;
|
||||||
|
import java.util.concurrent.atomic.AtomicReference;
|
||||||
|
|
||||||
|
import org.eclipse.jetty.io.IOFuture.Callback;
|
||||||
|
import org.junit.Assert;
|
||||||
|
import org.junit.Test;
|
||||||
|
|
||||||
|
public class IOFutureTest
|
||||||
|
{
|
||||||
|
@Test
|
||||||
|
public void testReadyCompleted() throws Exception
|
||||||
|
{
|
||||||
|
IOFuture future = new CompleteIOFuture();
|
||||||
|
|
||||||
|
assertTrue(future.isComplete());
|
||||||
|
assertTrue(future.isReady());
|
||||||
|
|
||||||
|
long start=System.currentTimeMillis();
|
||||||
|
future.block();
|
||||||
|
Assert.assertThat(System.currentTimeMillis()-start,lessThan(10L));
|
||||||
|
|
||||||
|
start=System.currentTimeMillis();
|
||||||
|
future.block(1000,TimeUnit.MILLISECONDS);
|
||||||
|
Assert.assertThat(System.currentTimeMillis()-start,lessThan(10L));
|
||||||
|
|
||||||
|
final AtomicBoolean ready = new AtomicBoolean(false);
|
||||||
|
final AtomicReference<Throwable> fail = new AtomicReference<>();
|
||||||
|
|
||||||
|
future.setCallback(new Callback()
|
||||||
|
{
|
||||||
|
@Override
|
||||||
|
public void onReady()
|
||||||
|
{
|
||||||
|
ready.set(true);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onFail(Throwable cause)
|
||||||
|
{
|
||||||
|
fail.set(cause);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
assertTrue(ready.get());
|
||||||
|
assertEquals((Throwable)null,fail.get());
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testFailedCompleted() throws Exception
|
||||||
|
{
|
||||||
|
Exception ex=new Exception("failed");
|
||||||
|
IOFuture future = new CompleteIOFuture(ex);
|
||||||
|
|
||||||
|
assertTrue(future.isComplete());
|
||||||
|
try
|
||||||
|
{
|
||||||
|
future.isReady();
|
||||||
|
Assert.fail();
|
||||||
|
}
|
||||||
|
catch(ExecutionException e)
|
||||||
|
{
|
||||||
|
Assert.assertEquals(ex,e.getCause());
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
long start=System.currentTimeMillis();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
future.block();
|
||||||
|
Assert.fail();
|
||||||
|
}
|
||||||
|
catch(ExecutionException e)
|
||||||
|
{
|
||||||
|
Assert.assertEquals(ex,e.getCause());
|
||||||
|
}
|
||||||
|
Assert.assertThat(System.currentTimeMillis()-start,lessThan(10L));
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
start=System.currentTimeMillis();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
future.block(1000,TimeUnit.MILLISECONDS);
|
||||||
|
Assert.fail();
|
||||||
|
}
|
||||||
|
catch(ExecutionException e)
|
||||||
|
{
|
||||||
|
Assert.assertEquals(ex,e.getCause());
|
||||||
|
}
|
||||||
|
Assert.assertThat(System.currentTimeMillis()-start,lessThan(10L));
|
||||||
|
|
||||||
|
|
||||||
|
final AtomicBoolean ready = new AtomicBoolean(false);
|
||||||
|
final AtomicReference<Throwable> fail = new AtomicReference<>();
|
||||||
|
|
||||||
|
future.setCallback(new Callback()
|
||||||
|
{
|
||||||
|
@Override
|
||||||
|
public void onReady()
|
||||||
|
{
|
||||||
|
ready.set(true);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onFail(Throwable cause)
|
||||||
|
{
|
||||||
|
fail.set(cause);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
assertFalse(ready.get());
|
||||||
|
assertEquals(ex,fail.get());
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testInCompleted() throws Exception
|
||||||
|
{
|
||||||
|
IOFuture future = new RecycledIOFuture();
|
||||||
|
|
||||||
|
assertFalse(future.isComplete());
|
||||||
|
assertFalse(future.isReady());
|
||||||
|
|
||||||
|
long start=System.currentTimeMillis();
|
||||||
|
future.block(10,TimeUnit.MILLISECONDS);
|
||||||
|
Assert.assertThat(System.currentTimeMillis()-start,greaterThan(9L));
|
||||||
|
|
||||||
|
final AtomicBoolean ready = new AtomicBoolean(false);
|
||||||
|
final AtomicReference<Throwable> fail = new AtomicReference<>();
|
||||||
|
|
||||||
|
future.setCallback(new Callback()
|
||||||
|
{
|
||||||
|
@Override
|
||||||
|
public void onReady()
|
||||||
|
{
|
||||||
|
ready.set(true);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onFail(Throwable cause)
|
||||||
|
{
|
||||||
|
fail.set(cause);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
assertFalse(ready.get());
|
||||||
|
assertEquals((Throwable)null,fail.get());
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testReady() throws Exception
|
||||||
|
{
|
||||||
|
final RecycledIOFuture future = new RecycledIOFuture();
|
||||||
|
|
||||||
|
assertFalse(future.isComplete());
|
||||||
|
assertFalse(future.isReady());
|
||||||
|
|
||||||
|
final AtomicBoolean ready = new AtomicBoolean(false);
|
||||||
|
final AtomicReference<Throwable> fail = new AtomicReference<>();
|
||||||
|
|
||||||
|
future.setCallback(new Callback()
|
||||||
|
{
|
||||||
|
@Override
|
||||||
|
public void onReady()
|
||||||
|
{
|
||||||
|
ready.set(true);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onFail(Throwable cause)
|
||||||
|
{
|
||||||
|
fail.set(cause);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
long start=System.currentTimeMillis();
|
||||||
|
assertFalse(future.block(10,TimeUnit.MILLISECONDS));
|
||||||
|
assertThat(System.currentTimeMillis()-start,greaterThan(9L));
|
||||||
|
|
||||||
|
assertFalse(ready.get());
|
||||||
|
assertEquals((Throwable)null,fail.get());
|
||||||
|
|
||||||
|
|
||||||
|
start=System.currentTimeMillis();
|
||||||
|
new Thread()
|
||||||
|
{
|
||||||
|
@Override
|
||||||
|
public void run()
|
||||||
|
{
|
||||||
|
try{TimeUnit.MILLISECONDS.sleep(50);}catch(Exception e){}
|
||||||
|
future.ready();
|
||||||
|
}
|
||||||
|
}.start();
|
||||||
|
|
||||||
|
assertTrue(future.block(1000,TimeUnit.MILLISECONDS));
|
||||||
|
Assert.assertThat(System.currentTimeMillis()-start,greaterThan(49L));
|
||||||
|
Assert.assertThat(System.currentTimeMillis()-start,lessThan(1000L));
|
||||||
|
|
||||||
|
assertTrue(future.isComplete());
|
||||||
|
assertTrue(future.isReady());
|
||||||
|
assertTrue(ready.get());
|
||||||
|
assertEquals((Throwable)null,fail.get());
|
||||||
|
|
||||||
|
ready.set(false);
|
||||||
|
future.recycle();
|
||||||
|
assertFalse(future.isComplete());
|
||||||
|
assertFalse(future.isReady());
|
||||||
|
start=System.currentTimeMillis();
|
||||||
|
new Thread()
|
||||||
|
{
|
||||||
|
@Override
|
||||||
|
public void run()
|
||||||
|
{
|
||||||
|
try{TimeUnit.MILLISECONDS.sleep(50);}catch(Exception e){}
|
||||||
|
future.ready();
|
||||||
|
}
|
||||||
|
}.start();
|
||||||
|
|
||||||
|
future.block();
|
||||||
|
Assert.assertThat(System.currentTimeMillis()-start,greaterThan(49L));
|
||||||
|
|
||||||
|
assertTrue(future.isComplete());
|
||||||
|
assertTrue(future.isReady());
|
||||||
|
assertFalse(ready.get()); // no callback set
|
||||||
|
assertEquals((Throwable)null,fail.get());
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testFail() throws Exception
|
||||||
|
{
|
||||||
|
final RecycledIOFuture future = new RecycledIOFuture();
|
||||||
|
final Exception ex=new Exception("failed");
|
||||||
|
|
||||||
|
assertFalse(future.isComplete());
|
||||||
|
assertFalse(future.isReady());
|
||||||
|
|
||||||
|
final AtomicBoolean ready = new AtomicBoolean(false);
|
||||||
|
final AtomicReference<Throwable> fail = new AtomicReference<>();
|
||||||
|
|
||||||
|
future.setCallback(new Callback()
|
||||||
|
{
|
||||||
|
@Override
|
||||||
|
public void onReady()
|
||||||
|
{
|
||||||
|
ready.set(true);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onFail(Throwable cause)
|
||||||
|
{
|
||||||
|
fail.set(cause);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
long start=System.currentTimeMillis();
|
||||||
|
assertFalse(future.block(10,TimeUnit.MILLISECONDS));
|
||||||
|
assertThat(System.currentTimeMillis()-start,greaterThan(9L));
|
||||||
|
|
||||||
|
assertFalse(ready.get());
|
||||||
|
assertEquals((Throwable)null,fail.get());
|
||||||
|
|
||||||
|
start=System.currentTimeMillis();
|
||||||
|
new Thread()
|
||||||
|
{
|
||||||
|
@Override
|
||||||
|
public void run()
|
||||||
|
{
|
||||||
|
try{TimeUnit.MILLISECONDS.sleep(50);}catch(Exception e){}
|
||||||
|
future.fail(ex);
|
||||||
|
}
|
||||||
|
}.start();
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
future.block(1000,TimeUnit.MILLISECONDS);
|
||||||
|
Assert.fail();
|
||||||
|
}
|
||||||
|
catch(ExecutionException e)
|
||||||
|
{
|
||||||
|
Assert.assertEquals(ex,e.getCause());
|
||||||
|
}
|
||||||
|
Assert.assertThat(System.currentTimeMillis()-start,greaterThan(49L));
|
||||||
|
Assert.assertThat(System.currentTimeMillis()-start,lessThan(1000L));
|
||||||
|
|
||||||
|
assertTrue(future.isComplete());
|
||||||
|
try
|
||||||
|
{
|
||||||
|
future.isReady();
|
||||||
|
Assert.fail();
|
||||||
|
}
|
||||||
|
catch(ExecutionException e)
|
||||||
|
{
|
||||||
|
Assert.assertEquals(ex,e.getCause());
|
||||||
|
}
|
||||||
|
assertFalse(ready.get());
|
||||||
|
assertEquals(ex,fail.get());
|
||||||
|
|
||||||
|
ready.set(false);
|
||||||
|
fail.set(null);
|
||||||
|
future.recycle();
|
||||||
|
assertFalse(future.isComplete());
|
||||||
|
assertFalse(future.isReady());
|
||||||
|
start=System.currentTimeMillis();
|
||||||
|
new Thread()
|
||||||
|
{
|
||||||
|
@Override
|
||||||
|
public void run()
|
||||||
|
{
|
||||||
|
try{TimeUnit.MILLISECONDS.sleep(50);}catch(Exception e){}
|
||||||
|
future.fail(ex);
|
||||||
|
}
|
||||||
|
}.start();
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
future.block();
|
||||||
|
Assert.fail();
|
||||||
|
}
|
||||||
|
catch(ExecutionException e)
|
||||||
|
{
|
||||||
|
Assert.assertEquals(ex,e.getCause());
|
||||||
|
}
|
||||||
|
Assert.assertThat(System.currentTimeMillis()-start,greaterThan(49L));
|
||||||
|
|
||||||
|
assertTrue(future.isComplete());
|
||||||
|
try
|
||||||
|
{
|
||||||
|
future.isReady();
|
||||||
|
Assert.fail();
|
||||||
|
}
|
||||||
|
catch(ExecutionException e)
|
||||||
|
{
|
||||||
|
Assert.assertEquals(ex,e.getCause());
|
||||||
|
}
|
||||||
|
assertFalse(ready.get()); // no callback set
|
||||||
|
assertEquals((Throwable)null,fail.get()); // no callback set
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
Loading…
Reference in New Issue