diff --git a/spring-web-reactive/src/main/java/org/springframework/rx/util/AbstractUnicastAsyncSubscriber.java b/spring-web-reactive/src/main/java/org/springframework/rx/util/AbstractUnicastAsyncSubscriber.java deleted file mode 100644 index aece160be8..0000000000 --- a/spring-web-reactive/src/main/java/org/springframework/rx/util/AbstractUnicastAsyncSubscriber.java +++ /dev/null @@ -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 implements Subscriber { - - 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(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> inboundSignals = - new ConcurrentLinkedQueue>(); - - 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 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(); - } - } - } - - } - } -} \ No newline at end of file diff --git a/spring-web-reactive/src/main/java/org/springframework/rx/util/AbstractUnicastSyncSubscriber.java b/spring-web-reactive/src/main/java/org/springframework/rx/util/AbstractUnicastSyncSubscriber.java deleted file mode 100644 index 85e4764d50..0000000000 --- a/spring-web-reactive/src/main/java/org/springframework/rx/util/AbstractUnicastSyncSubscriber.java +++ /dev/null @@ -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 implements Subscriber { - - 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; - - -} diff --git a/spring-web-reactive/src/main/java/org/springframework/rx/util/BlockingSignalQueue.java b/spring-web-reactive/src/main/java/org/springframework/rx/util/BlockingSignalQueue.java index f30f7470d4..476ab16eb4 100644 --- a/spring-web-reactive/src/main/java/org/springframework/rx/util/BlockingSignalQueue.java +++ b/spring-web-reactive/src/main/java/org/springframework/rx/util/BlockingSignalQueue.java @@ -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. * *

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 { 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)); } /** @@ -59,12 +57,13 @@ public class BlockingSignalQueue { 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(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 { */ public boolean isHeadSignal() { Signal signal = this.queue.peek(); - return signal instanceof OnNext; + return signal != null && signal.isOnNext(); } /** @@ -84,7 +83,7 @@ public class BlockingSignalQueue { */ public boolean isHeadError() { Signal signal = this.queue.peek(); - return signal instanceof OnError; + return signal != null && signal.isOnError(); } /** @@ -93,7 +92,7 @@ public class BlockingSignalQueue { */ public boolean isComplete() { Signal signal = this.queue.peek(); - return OnComplete.INSTANCE == signal; + return signal != null && signal.isComplete(); } /** @@ -120,4 +119,121 @@ public class BlockingSignalQueue { return signal != null ? signal.error() : null; } + private interface Signal { + + boolean isOnNext(); + + T next(); + + boolean isOnError(); + + Throwable error(); + + boolean isComplete(); + } + + private static class OnNext implements Signal { + + 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 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 T next() { + throw new IllegalStateException(); + } + + @Override + public boolean isComplete() { + return false; + } + } + + private static class OnComplete implements Signal { + + 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(); + } + + } + } diff --git a/spring-web-reactive/src/main/java/org/springframework/rx/util/OnComplete.java b/spring-web-reactive/src/main/java/org/springframework/rx/util/OnComplete.java deleted file mode 100644 index c306e584cf..0000000000 --- a/spring-web-reactive/src/main/java/org/springframework/rx/util/OnComplete.java +++ /dev/null @@ -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(); - } -} diff --git a/spring-web-reactive/src/main/java/org/springframework/rx/util/OnError.java b/spring-web-reactive/src/main/java/org/springframework/rx/util/OnError.java deleted file mode 100644 index 0135485932..0000000000 --- a/spring-web-reactive/src/main/java/org/springframework/rx/util/OnError.java +++ /dev/null @@ -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; - } -} diff --git a/spring-web-reactive/src/main/java/org/springframework/rx/util/OnNext.java b/spring-web-reactive/src/main/java/org/springframework/rx/util/OnNext.java deleted file mode 100644 index 56e052e1aa..0000000000 --- a/spring-web-reactive/src/main/java/org/springframework/rx/util/OnNext.java +++ /dev/null @@ -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 implements Signal { - - 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; - } -} diff --git a/spring-web-reactive/src/main/java/org/springframework/rx/util/OnSubscribe.java b/spring-web-reactive/src/main/java/org/springframework/rx/util/OnSubscribe.java deleted file mode 100644 index d4731a403b..0000000000 --- a/spring-web-reactive/src/main/java/org/springframework/rx/util/OnSubscribe.java +++ /dev/null @@ -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; - } -} diff --git a/spring-web-reactive/src/main/java/org/springframework/rx/util/Signal.java b/spring-web-reactive/src/main/java/org/springframework/rx/util/Signal.java deleted file mode 100644 index 1833ef210e..0000000000 --- a/spring-web-reactive/src/main/java/org/springframework/rx/util/Signal.java +++ /dev/null @@ -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 { - - boolean isOnNext(); - - T next(); - - boolean isOnError(); - - Throwable error(); - - boolean isOnSubscribe(); - - Subscription subscription(); - - boolean isComplete(); -} diff --git a/spring-web-reactive/src/test/java/org/springframework/rx/util/BlockingByteBufQueuePublisherTests.java b/spring-web-reactive/src/test/java/org/springframework/rx/util/BlockingByteBufQueuePublisherTests.java index 5c14966818..ea06be0397 100644 --- a/spring-web-reactive/src/test/java/org/springframework/rx/util/BlockingByteBufQueuePublisherTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/rx/util/BlockingByteBufQueuePublisherTests.java @@ -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 queue; - private BlockingSignalQueuePublisher publisher; + private BlockingSignalQueuePublisher publisher; @Before public void setUp() throws Exception { - queue = new BlockingSignalQueue(); - publisher = new BlockingSignalQueuePublisher(queue); + queue = new BlockingSignalQueue(); + publisher = new BlockingSignalQueuePublisher(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 received = new ArrayList(2); + final List received = new ArrayList(2); + + publisher.subscribe(new Subscriber() { + private Subscription subscription; - publisher.subscribe(new Subscriber() { @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 received = new ArrayList(2); + final List received = new ArrayList(2); - publisher.subscribe(new Subscriber() { + publisher.subscribe(new Subscriber() { @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() { + publisher.subscribe(new Subscriber() { @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() { + publisher.subscribe(new Subscriber() { @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 received = new ArrayList(1); + + publisher.subscribe(new Subscriber() { + + 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)); } + } \ No newline at end of file diff --git a/spring-web-reactive/src/test/java/org/springframework/rx/util/BlockingByteBufQueueTests.java b/spring-web-reactive/src/test/java/org/springframework/rx/util/BlockingByteBufQueueTests.java index 882c6a13c2..3148d7df56 100644 --- a/spring-web-reactive/src/test/java/org/springframework/rx/util/BlockingByteBufQueueTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/rx/util/BlockingByteBufQueueTests.java @@ -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 queue; @Before public void setUp() throws Exception { - queue = new BlockingSignalQueue(); + queue = new BlockingSignalQueue(); } @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);