Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

work, work.

Browse Source
This commit is contained in:
Arjen Poutsma
2015-07-01 15:03:14 +02:00
parent c552aaa6f1
commit 3b45087c87
10 changed files with 205 additions and 644 deletions
Download Patch File Download Diff File Expand all files Collapse all files

214
spring-web-reactive/src/main/java/org/springframework/rx/util/AbstractUnicastAsyncSubscriber.java
View File

@@ -1,214 +0,0 @@
/*
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.rx.util;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.springframework.util.Assert;
/**
* @author Arjen Poutsma
*/
public abstract class AbstractUnicastAsyncSubscriber<T> implements Subscriber<T> {
private final Executor executor;
private Subscription subscription;
private boolean done;
protected AbstractUnicastAsyncSubscriber(Executor executor) {
Assert.notNull(executor, "'executor' must not be null");
this.executor = executor;
}
private void done() {
done = true;
if (subscription != null) {
subscription.cancel();
}
}
// This method is invoked when the OnNext signals arrive
// Returns whether more elements are desired or not, and if no more elements are desired,
// for convenience.
protected abstract boolean whenNext(final T element);
// This method is invoked when the OnComplete signal arrives
// override this method to implement your own custom onComplete logic.
protected void whenComplete() {
}
// This method is invoked if the OnError signal arrives
// override this method to implement your own custom onError logic.
protected void whenError(Throwable error) {
}
private void handleOnSubscribe(Subscription subscription) {
if (subscription == null) {
return;
}
if (this.subscription != null) {
subscription.cancel();
}
else {
this.subscription = subscription;
this.subscription.request(1);
}
}
private void handleOnNext(final T element) {
if (!done) {
try {
if (whenNext(element)) {
subscription.request(1);
}
else {
done();
}
}
catch (final Throwable t) {
done();
onError(t);
}
}
}
private void handleOnComplete() {
done = true;
whenComplete();
}
private void handleOnError(final Throwable error) {
done = true;
whenError(error);
}
// We implement the OnX methods on `Subscriber` to send Signals that we will process asycnhronously, but only one at a time
@Override
public final void onSubscribe(final Subscription s) {
// As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `Subscription` is `null`
if (s == null) {
throw null;
}
signal(new OnSubscribe(s));
}
@Override
public final void onNext(final T element) {
// As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `element` is `null`
if (element == null) {
throw null;
}
signal(new OnNext<T>(element));
}
@Override
public final void onError(final Throwable t) {
// As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `Throwable` is `null`
if (t == null) {
throw null;
}
signal(new OnError(t));
}
@Override
public final void onComplete() {
signal(OnComplete.INSTANCE);
}
private final ConcurrentLinkedQueue<Signal<T>> inboundSignals =
new ConcurrentLinkedQueue<Signal<T>>();
private final AtomicBoolean enabled = new AtomicBoolean(false);
// What `signal` does is that it sends signals to the `Subscription` asynchronously
private void signal(final Signal signal) {
if (inboundSignals
.offer(signal)) // No need to null-check here as ConcurrentLinkedQueue does this for us
{
tryScheduleToExecute(); // Then we try to schedule it for execution, if it isn't already
}
}
// This method makes sure that this `Subscriber` is only executing on one Thread at a time
private void tryScheduleToExecute() {
if (enabled.compareAndSet(false, true)) {
try {
executor.execute(new SignalRunnable());
}
catch (Throwable t) { // If we can't run on the `Executor`, we need to fail gracefully and not violate rule 2.13
if (!done) {
try {
done(); // First of all, this failure is not recoverable, so we need to cancel our subscription
}
finally {
inboundSignals.clear(); // We're not going to need these anymore
// This subscription is cancelled by now, but letting the Subscriber become schedulable again means
// that we can drain the inboundSignals queue if anything arrives after clearing
enabled.set(false);
}
}
}
}
}
private class SignalRunnable implements Runnable {
@Override
public void run() {
if (enabled.get()) {
try {
Signal<T> s = inboundSignals.poll();
if (!done) {
if (s.isOnNext()) {
handleOnNext(s.next());
}
else if (s.isOnSubscribe()) {
handleOnSubscribe(s.subscription());
}
else if (s.isOnError()) {
handleOnError(s.error());
}
else if (s.isComplete()) {
handleOnComplete();
}
}
}
finally {
enabled.set(false);
if (!inboundSignals.isEmpty()) {
tryScheduleToExecute();
}
}
}
}
}
}

76
spring-web-reactive/src/main/java/org/springframework/rx/util/AbstractUnicastSyncSubscriber.java
View File

@@ -1,76 +0,0 @@
/*
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.rx.util;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
/**
* @author Arjen Poutsma
*/
public abstract class AbstractUnicastSyncSubscriber<T> implements Subscriber<T> {
private Subscription subscription;
private boolean done = false;
@Override
public final void onSubscribe(Subscription subscription) {
if (subscription == null) {
throw new NullPointerException();
}
if (this.subscription != null) {
subscription.cancel();
}
else {
this.subscription = subscription;
this.subscription.request(1);
}
}
@Override
public final void onNext(T element) {
if (element == null) {
throw new NullPointerException();
}
if (!done) {
try {
if (onNextInternal(element)) {
subscription.request(1);
}
else {
done();
}
}
catch (Throwable t) {
done();
onError(t);
}
}
}
private void done() {
done = true;
subscription.cancel();
}
protected abstract boolean onNextInternal(final T element) throws Exception;
}

134
spring-web-reactive/src/main/java/org/springframework/rx/util/BlockingSignalQueue.java
View File

@@ -19,8 +19,6 @@ package org.springframework.rx.util;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import io.netty.buffer.ByteBuf;
import org.springframework.util.Assert;
/**
@@ -29,8 +27,8 @@ import org.springframework.util.Assert;
* streams.
*
* <p>Typically, this class will be used by two threads: one thread to put new elements on
* the stack by calling {@link #put(ByteBuf)}, possibly {@link #putError(Throwable)} and
* finally {@link #complete()}. The other thread will read elements by calling {@link
* the stack by calling {@link #putSignal(Object)}, possibly {@link #putError(Throwable)}
* and finally {@link #complete()}. The other thread will read elements by calling {@link
* #isHeadSignal()}/{@link #pollSignal()} and {@link #isHeadError()}/{@link #pollError()},
* while keeping an eye on {@link #isComplete()}.
* @author Arjen Poutsma
@@ -48,7 +46,7 @@ public class BlockingSignalQueue<T> {
public void putSignal(T t) throws InterruptedException {
Assert.notNull(t, "'t' must not be null");
Assert.state(!isComplete(), "Cannot put signal in queue after complete()");
this.queue.put(new OnNext(t));
this.queue.put(new OnNext<T>(t));
}
/**
@@ -59,12 +57,13 @@ public class BlockingSignalQueue<T> {
public void putError(Throwable error) throws InterruptedException {
Assert.notNull(error, "'error' must not be null");
Assert.state(!isComplete(), "Cannot putSignal errors in queue after complete()");
this.queue.put(new OnError(error));
this.queue.put(new OnError<T>(error));
}
/**
* Marks the queue as complete.
*/
@SuppressWarnings("unchecked")
public void complete() throws InterruptedException {
this.queue.put(OnComplete.INSTANCE);
}
@@ -75,7 +74,7 @@ public class BlockingSignalQueue<T> {
*/
public boolean isHeadSignal() {
Signal signal = this.queue.peek();
return signal instanceof OnNext;
return signal != null && signal.isOnNext();
}
/**
@@ -84,7 +83,7 @@ public class BlockingSignalQueue<T> {
*/
public boolean isHeadError() {
Signal signal = this.queue.peek();
return signal instanceof OnError;
return signal != null && signal.isOnError();
}
/**
@@ -93,7 +92,7 @@ public class BlockingSignalQueue<T> {
*/
public boolean isComplete() {
Signal signal = this.queue.peek();
return OnComplete.INSTANCE == signal;
return signal != null && signal.isComplete();
}
/**
@@ -120,4 +119,121 @@ public class BlockingSignalQueue<T> {
return signal != null ? signal.error() : null;
}
private interface Signal<T> {
boolean isOnNext();
T next();
boolean isOnError();
Throwable error();
boolean isComplete();
}
private static class OnNext<T> implements Signal<T> {
private final T next;
public OnNext(T next) {
Assert.notNull(next, "'next' must not be null");
this.next = next;
}
@Override
public boolean isOnNext() {
return true;
}
@Override
public T next() {
return next;
}
@Override
public boolean isOnError() {
return false;
}
@Override
public Throwable error() {
throw new IllegalStateException();
}
@Override
public boolean isComplete() {
return false;
}
}
private static final class OnError<T> implements Signal<T> {
private final Throwable error;
public OnError(Throwable error) {
Assert.notNull(error, "'error' must not be null");
this.error = error;
}
@Override
public boolean isOnError() {
return true;
}
@Override
public Throwable error() {
return error;
}
@Override
public boolean isOnNext() {
return false;
}
@Override
public T next() {
throw new IllegalStateException();
}
@Override
public boolean isComplete() {
return false;
}
}
private static class OnComplete<T> implements Signal<T> {
private static final OnComplete INSTANCE = new OnComplete();
private OnComplete() {
}
@Override
public boolean isComplete() {
return true;
}
@Override
public boolean isOnNext() {
return false;
}
@Override
public T next() {
throw new IllegalStateException();
}
@Override
public boolean isOnError() {
return false;
}
@Override
public Throwable error() {
throw new IllegalStateException();
}
}
}

65
spring-web-reactive/src/main/java/org/springframework/rx/util/OnComplete.java
View File

@@ -1,65 +0,0 @@
/*
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.rx.util;
import org.reactivestreams.Subscription;
/**
* @author Arjen Poutsma
*/
class OnComplete implements Signal {
public static final OnComplete INSTANCE = new OnComplete();
private OnComplete() {
}
@Override
public boolean isComplete() {
return true;
}
@Override
public boolean isOnNext() {
return false;
}
@Override
public Object next() {
throw new IllegalStateException();
}
@Override
public boolean isOnError() {
return false;
}
@Override
public Throwable error() {
throw new IllegalStateException();
}
@Override
public boolean isOnSubscribe() {
return false;
}
@Override
public Subscription subscription() {
throw new IllegalStateException();
}
}

69
spring-web-reactive/src/main/java/org/springframework/rx/util/OnError.java
View File

@@ -1,69 +0,0 @@
/*
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.rx.util;
import org.reactivestreams.Subscription;
import org.springframework.util.Assert;
/**
* @author Arjen Poutsma
*/
final class OnError implements Signal {
private final Throwable error;
public OnError(Throwable error) {
Assert.notNull(error, "'error' must not be null");
this.error = error;
}
@Override
public boolean isOnError() {
return true;
}
@Override
public Throwable error() {
return error;
}
@Override
public boolean isOnNext() {
return false;
}
@Override
public Object next() {
throw new IllegalStateException();
}
@Override
public boolean isOnSubscribe() {
return false;
}
@Override
public Subscription subscription() {
throw new IllegalStateException();
}
@Override
public boolean isComplete() {
return false;
}
}

70
spring-web-reactive/src/main/java/org/springframework/rx/util/OnNext.java
View File

@@ -1,70 +0,0 @@
/*
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.rx.util;
import org.reactivestreams.Subscription;
import org.springframework.util.Assert;
/**
* @author Arjen Poutsma
*/
class OnNext<T> implements Signal<T> {
private final T next;
public OnNext(T next) {
Assert.notNull(next, "'next' must not be null");
this.next = next;
}
@Override
public boolean isOnNext() {
return true;
}
@Override
public T next() {
return next;
}
@Override
public boolean isOnError() {
return false;
}
@Override
public Throwable error() {
throw new IllegalStateException();
}
@Override
public boolean isOnSubscribe() {
return false;
}
@Override
public Subscription subscription() {
throw new IllegalStateException();
}
@Override
public boolean isComplete() {
return false;
}
}

70
spring-web-reactive/src/main/java/org/springframework/rx/util/OnSubscribe.java
View File

@@ -1,70 +0,0 @@
/*
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.rx.util;
import org.reactivestreams.Subscription;
import org.springframework.util.Assert;
/**
* @author Arjen Poutsma
*/
class OnSubscribe implements Signal {
private final Subscription subscription;
public OnSubscribe(Subscription subscription) {
Assert.notNull(subscription, "'subscription' must not be null");
this.subscription = subscription;
}
@Override
public boolean isOnSubscribe() {
return true;
}
@Override
public Subscription subscription() {
return null;
}
@Override
public boolean isOnNext() {
return false;
}
@Override
public Object next() {
throw new IllegalStateException();
}
@Override
public boolean isOnError() {
return false;
}
@Override
public Throwable error() {
throw new IllegalStateException();
}
@Override
public boolean isComplete() {
return false;
}
}

39
spring-web-reactive/src/main/java/org/springframework/rx/util/Signal.java
View File

@@ -1,39 +0,0 @@
/*
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.rx.util;
import org.reactivestreams.Subscription;
/**
* @author Arjen Poutsma
*/
interface Signal<T> {
boolean isOnNext();
T next();
boolean isOnError();
Throwable error();
boolean isOnSubscribe();
Subscription subscription();
boolean isComplete();
}

97
spring-web-reactive/src/test/java/org/springframework/rx/util/BlockingByteBufQueuePublisherTests.java
View File

@@ -20,50 +20,53 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import static org.junit.Assert.*;
/**
* @author Arjen Poutsma
*/
public class BlockingByteBufQueuePublisherTests {
private BlockingSignalQueue queue;
private BlockingSignalQueue<byte[]> queue;
private BlockingSignalQueuePublisher publisher;
private BlockingSignalQueuePublisher<byte[]> publisher;
@Before
public void setUp() throws Exception {
queue = new BlockingSignalQueue();
publisher = new BlockingSignalQueuePublisher(queue);
queue = new BlockingSignalQueue<byte[]>();
publisher = new BlockingSignalQueuePublisher<byte[]>(queue);
}
@Test
public void normal() throws Exception {
ByteBuf abc = Unpooled.copiedBuffer(new byte[]{'a', 'b', 'c'});
ByteBuf def = Unpooled.copiedBuffer(new byte[]{'d', 'e', 'f'});
byte[] abc = new byte[]{'a', 'b', 'c'};
byte[] def = new byte[]{'d', 'e', 'f'};
queue.putSignal(abc);
queue.putSignal(def);
queue.complete();
final AtomicBoolean complete = new AtomicBoolean(false);
final List<ByteBuf> received = new ArrayList<ByteBuf>(2);
final List<byte[]> received = new ArrayList<byte[]>(2);
publisher.subscribe(new Subscriber<byte[]>() {
private Subscription subscription;
publisher.subscribe(new Subscriber<ByteBuf>() {
@Override
public void onSubscribe(Subscription s) {
s.request(2);
s.request(1);
this.subscription = s;
}
@Override
public void onNext(ByteBuf byteBuf) {
received.add(byteBuf);
public void onNext(byte[] bytes) {
received.add(bytes);
this.subscription.request(1);
}
@Override
@@ -87,25 +90,25 @@ public class BlockingByteBufQueuePublisherTests {
@Test
public void unbounded() throws Exception {
ByteBuf abc = Unpooled.copiedBuffer(new byte[]{'a', 'b', 'c'});
ByteBuf def = Unpooled.copiedBuffer(new byte[]{'d', 'e', 'f'});
byte[] abc = new byte[]{'a', 'b', 'c'};
byte[] def = new byte[]{'d', 'e', 'f'};
queue.putSignal(abc);
queue.putSignal(def);
queue.complete();
final AtomicBoolean complete = new AtomicBoolean(false);
final List<ByteBuf> received = new ArrayList<ByteBuf>(2);
final List<byte[]> received = new ArrayList<byte[]>(2);
publisher.subscribe(new Subscriber<ByteBuf>() {
publisher.subscribe(new Subscriber<byte[]>() {
@Override
public void onSubscribe(Subscription s) {
s.request(Long.MAX_VALUE);
}
@Override
public void onNext(ByteBuf byteBuf) {
received.add(byteBuf);
public void onNext(byte[] bytes) {
received.add(bytes);
}
@Override
@@ -129,14 +132,14 @@ public class BlockingByteBufQueuePublisherTests {
@Test
public void multipleSubscribe() throws Exception {
publisher.subscribe(new Subscriber<ByteBuf>() {
publisher.subscribe(new Subscriber<byte[]>() {
@Override
public void onSubscribe(Subscription s) {
}
@Override
public void onNext(ByteBuf byteBuf) {
public void onNext(byte[] bytes) {
}
@@ -150,14 +153,14 @@ public class BlockingByteBufQueuePublisherTests {
}
});
publisher.subscribe(new Subscriber<ByteBuf>() {
publisher.subscribe(new Subscriber<byte[]>() {
@Override
public void onSubscribe(Subscription s) {
fail("onSubscribe not expected");
}
@Override
public void onNext(ByteBuf byteBuf) {
public void onNext(byte[] bytes) {
fail("onNext not expected");
}
@@ -171,8 +174,54 @@ public class BlockingByteBufQueuePublisherTests {
fail("onComplete not expected");
}
});
}
@Test
public void cancel() throws Exception {
byte[] abc = new byte[]{'a', 'b', 'c'};
byte[] def = new byte[]{'d', 'e', 'f'};
queue.putSignal(abc);
queue.putSignal(def);
queue.complete();
final AtomicBoolean complete = new AtomicBoolean(false);
final List<byte[]> received = new ArrayList<byte[]>(1);
publisher.subscribe(new Subscriber<byte[]>() {
private Subscription subscription;
@Override
public void onSubscribe(Subscription s) {
s.request(1);
this.subscription = s;
}
@Override
public void onNext(byte[] bytes) {
received.add(bytes);
this.subscription.cancel();
complete.set(true);
}
@Override
public void onError(Throwable t) {
fail("onError not expected");
}
@Override
public void onComplete() {
}
});
while (!complete.get()) {
}
assertEquals(1, received.size());
assertSame(abc, received.get(0));
}
}

15
spring-web-reactive/src/test/java/org/springframework/rx/util/BlockingByteBufQueueTests.java
View File

@@ -16,28 +16,27 @@
package org.springframework.rx.util;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.*;
/**
* @author Arjen Poutsma
*/
public class BlockingByteBufQueueTests {
private BlockingSignalQueue queue;
private BlockingSignalQueue<byte[]> queue;
@Before
public void setUp() throws Exception {
queue = new BlockingSignalQueue();
queue = new BlockingSignalQueue<byte[]>();
}
@Test
public void normal() throws Exception {
ByteBuf abc = Unpooled.copiedBuffer(new byte[]{'a', 'b', 'c'});
ByteBuf def = Unpooled.copiedBuffer(new byte[]{'d', 'e', 'f'});
byte[] abc = new byte[]{'a', 'b', 'c'};
byte[] def = new byte[]{'d', 'e', 'f'};
queue.putSignal(abc);
queue.putSignal(def);
@@ -57,7 +56,7 @@ public class BlockingByteBufQueueTests {
@Test
public void error() throws Exception {
ByteBuf abc = Unpooled.copiedBuffer(new byte[]{'a', 'b', 'c'});
byte[] abc = new byte[]{'a', 'b', 'c'};
Throwable error = new IllegalStateException();
queue.putSignal(abc);