diff --git a/CopyrightWaivers.txt b/CopyrightWaivers.txt index 4863b73b..1030e76d 100644 --- a/CopyrightWaivers.txt +++ b/CopyrightWaivers.txt @@ -41,3 +41,8 @@ patriknw | Patrik Nordwall, patrik.nordwall@gmail.com, Lightbend Inc angelsanz | Ángel Sanz, angelsanz@users.noreply.github.com shenghaiyang | 盛海洋, shenghaiyang@aliyun.com kiiadi | Kyle Thomson, kylthoms@amazon.com, Amazon.com +jroper | James Roper, james@jazzy.id.au, Lightbend Inc. +olegdokuka | Oleh Dokuka, shadowgun@.i.ua, Netifi Inc. +Scottmitch | Scott Mitchell, scott_mitchell@apple.com, Apple Inc. +retronym | Jason Zaugg, jzaugg@gmail.com, Lightbend Inc. + diff --git a/README.md b/README.md index faef51c9..1bf97923 100644 --- a/README.md +++ b/README.md @@ -8,12 +8,12 @@ The latest release is available on Maven Central as org.reactivestreams reactive-streams - 1.0.2-RC2 + 1.0.3 org.reactivestreams reactive-streams-tck - 1.0.2-RC2 + 1.0.3 test ``` @@ -79,12 +79,12 @@ followed by a possibly unbounded number of `onNext` signals (as requested by `Su | Non-obstructing | Quality describing a method which is as quick to execute as possible—on the calling thread. This means, for example, avoids heavy computations and other things that would stall the caller´s thread of execution. | | Terminal state | For a Publisher: When `onComplete` or `onError` has been signalled. For a Subscriber: When an `onComplete` or `onError` has been received.| | NOP | Execution that has no detectable effect to the calling thread, and can as such safely be called any number of times.| -| External synchronization | Access coordination for thread safety purposes implemented outside of the constructs defined in this specification, using techniques such as, but not limited to, `atomics`, `monitors`, or `locks`. | +| Serial(ly) | In the context of a [Signal](#term_signal), non-overlapping. In the context of the JVM, calls to methods on an object are serial if and only if there is a happens-before relationship between those calls (implying also that the calls do not overlap). When the calls are performed asynchronously, coordination to establish the happens-before relationship is to be implemented using techniques such as, but not limited to, atomics, monitors, or locks. | | Thread-safe | Can be safely invoked synchronously, or asychronously, without requiring external synchronization to ensure program correctness. | ### SPECIFICATION -#### 1. Publisher ([Code](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.2-RC2/api/src/main/java/org/reactivestreams/Publisher.java)) +#### 1. Publisher ([Code](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.3/api/src/main/java/org/reactivestreams/Publisher.java)) ```java public interface Publisher { @@ -98,8 +98,8 @@ public interface Publisher { | [:bulb:](#1.1 "1.1 explained") | *The intent of this rule is to make it clear that Publishers cannot signal more elements than Subscribers have requested. There’s an implicit, but important, consequence to this rule: Since demand can only be fulfilled after it has been received, there’s a happens-before relationship between requesting elements and receiving elements.* | | 2 | A `Publisher` MAY signal fewer `onNext` than requested and terminate the `Subscription` by calling `onComplete` or `onError`. | | [:bulb:](#1.2 "1.2 explained") | *The intent of this rule is to make it clear that a Publisher cannot guarantee that it will be able to produce the number of elements requested; it simply might not be able to produce them all; it may be in a failed state; it may be empty or otherwise already completed.* | -| 3 | `onSubscribe`, `onNext`, `onError` and `onComplete` signaled to a `Subscriber` MUST be signaled in a [thread-safe](#term_thread-safe) manner—and if performed by multiple threads—use [external synchronization](#term_ext_sync). | -| [:bulb:](#1.3 "1.3 explained") | *The intent of this rule is to make it clear that [external synchronization](#term_ext_sync) must be employed if the Publisher intends to send signals from multiple/different threads.* | +| 3 | `onSubscribe`, `onNext`, `onError` and `onComplete` signaled to a `Subscriber` MUST be signaled [serially](#term_serially). | +| [:bulb:](#1.3 "1.3 explained") | *The intent of this rule is to permit the signalling of signals (including from multiple threads) if and only if a happens-before relation between each of the signals is established.* | | 4 | If a `Publisher` fails it MUST signal an `onError`. | | [:bulb:](#1.4 "1.4 explained") | *The intent of this rule is to make it clear that a Publisher is responsible for notifying its Subscribers if it detects that it cannot proceed—Subscribers must be given a chance to clean up resources or otherwise deal with the Publisher´s failures.* | | 5 | If a `Publisher` terminates successfully (finite stream) it MUST signal an `onComplete`. | @@ -117,7 +117,7 @@ public interface Publisher { | 11 | A `Publisher` MAY support multiple `Subscriber`s and decides whether each `Subscription` is unicast or multicast. | | [:bulb:](#1.11 "1.11 explained") | *The intent of this rule is to give Publisher implementations the flexibility to decide how many, if any, Subscribers they will support, and how elements are going to be distributed.* | -#### 2. Subscriber ([Code](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.2-RC2/api/src/main/java/org/reactivestreams/Subscriber.java)) +#### 2. Subscriber ([Code](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.3/api/src/main/java/org/reactivestreams/Subscriber.java)) ```java public interface Subscriber { @@ -131,7 +131,7 @@ public interface Subscriber { | ID | Rule | | ------------------------- | ------------------------------------------------------------------------------------------------------ | | 1 | A `Subscriber` MUST signal demand via `Subscription.request(long n)` to receive `onNext` signals. | -| [:bulb:](#2.1 "2.1 explained") | *The intent of this rule is to establish that it is the responsibility of the Subscriber to signal when, and how many, elements it is able and willing to receive.* | +| [:bulb:](#2.1 "2.1 explained") | *The intent of this rule is to establish that it is the responsibility of the Subscriber to decide when and how many elements it is able and willing to receive. To avoid signal reordering caused by reentrant Subscription methods, it is strongly RECOMMENDED for synchronous Subscriber implementations to invoke Subscription methods at the very end of any signal processing. It is RECOMMENDED that Subscribers request the upper limit of what they are able to process, as requesting only one element at a time results in an inherently inefficient "stop-and-wait" protocol.* | | 2 | If a `Subscriber` suspects that its processing of signals will negatively impact its `Publisher`´s responsivity, it is RECOMMENDED that it asynchronously dispatches its signals. | | [:bulb:](#2.2 "2.2 explained") | *The intent of this rule is that a Subscriber should [not obstruct](#term_non-obstructing) the progress of the Publisher from an execution point-of-view. In other words, the Subscriber should not starve the Publisher from receiving CPU cycles.* | | 3 | `Subscriber.onComplete()` and `Subscriber.onError(Throwable t)` MUST NOT call any methods on the `Subscription` or the `Publisher`. | @@ -139,11 +139,11 @@ public interface Subscriber { | 4 | `Subscriber.onComplete()` and `Subscriber.onError(Throwable t)` MUST consider the Subscription cancelled after having received the signal. | | [:bulb:](#2.4 "2.4 explained") | *The intent of this rule is to make sure that Subscribers respect a Publisher’s [terminal state](#term_terminal_state) signals. A Subscription is simply not valid anymore after an onComplete or onError signal has been received.* | | 5 | A `Subscriber` MUST call `Subscription.cancel()` on the given `Subscription` after an `onSubscribe` signal if it already has an active `Subscription`. | -| [:bulb:](#2.5 "2.5 explained") | *The intent of this rule is to prevent that two, or more, separate Publishers from thinking that they can interact with the same Subscriber. Enforcing this rule means that resource leaks are prevented since extra Subscriptions will be cancelled.* | +| [:bulb:](#2.5 "2.5 explained") | *The intent of this rule is to prevent that two, or more, separate Publishers from trying to interact with the same Subscriber. Enforcing this rule means that resource leaks are prevented since extra Subscriptions will be cancelled. Failure to conform to this rule may lead to violations of Publisher rule 1, amongst others. Such violations can lead to hard-to-diagnose bugs.* | | 6 | A `Subscriber` MUST call `Subscription.cancel()` if the `Subscription` is no longer needed. | | [:bulb:](#2.6 "2.6 explained") | *The intent of this rule is to establish that Subscribers cannot just throw Subscriptions away when they are no longer needed, they have to call `cancel` so that resources held by that Subscription can be safely, and timely, reclaimed. An example of this would be a Subscriber which is only interested in a specific element, which would then cancel its Subscription to signal its completion to the Publisher.* | -| 7 | A `Subscriber` MUST ensure that all calls on its `Subscription` take place from the same thread or provide for respective [external synchronization](#term_ext_sync). | -| [:bulb:](#2.7 "2.7 explained") | *The intent of this rule is to establish that [external synchronization](#term_ext_sync) must be added if a Subscriber will be using a Subscription concurrently by two or more threads.* | +| 7 | A Subscriber MUST ensure that all calls on its Subscription's request and cancel methods are performed [serially](#term_serially). | +| [:bulb:](#2.7 "2.7 explained") | *The intent of this rule is to permit the calling of the request and cancel methods (including from multiple threads) if and only if a happens-before relation between each of the calls is established.* | | 8 | A `Subscriber` MUST be prepared to receive one or more `onNext` signals after having called `Subscription.cancel()` if there are still requested elements pending [see [3.12](#3.12)]. `Subscription.cancel()` does not guarantee to perform the underlying cleaning operations immediately. | | [:bulb:](#2.8 "2.8 explained") | *The intent of this rule is to highlight that there may be a delay between calling `cancel` and the Publisher observing that cancellation.* | | 9 | A `Subscriber` MUST be prepared to receive an `onComplete` signal with or without a preceding `Subscription.request(long n)` call. | @@ -157,7 +157,7 @@ public interface Subscriber { | 13 | Calling `onSubscribe`, `onNext`, `onError` or `onComplete` MUST [return normally](#term_return_normally) except when any provided parameter is `null` in which case it MUST throw a `java.lang.NullPointerException` to the caller, for all other situations the only legal way for a `Subscriber` to signal failure is by cancelling its `Subscription`. In the case that this rule is violated, any associated `Subscription` to the `Subscriber` MUST be considered as cancelled, and the caller MUST raise this error condition in a fashion that is adequate for the runtime environment. | | [:bulb:](#2.13 "2.13 explained") | *The intent of this rule is to establish the semantics for the methods of Subscriber and what the Publisher is allowed to do in which case this rule is violated. «Raise this error condition in a fashion that is adequate for the runtime environment» could mean logging the error—or otherwise make someone or something aware of the situation—as the error cannot be signalled to the faulty Subscriber.* | -#### 3. Subscription ([Code](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.2-RC2/api/src/main/java/org/reactivestreams/Subscription.java)) +#### 3. Subscription ([Code](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.3/api/src/main/java/org/reactivestreams/Subscription.java)) ```java public interface Subscription { @@ -205,7 +205,7 @@ public interface Subscription { A `Subscription` is shared by exactly one `Publisher` and one `Subscriber` for the purpose of mediating the data exchange between this pair. This is the reason why the `subscribe()` method does not return the created `Subscription`, but instead returns `void`; the `Subscription` is only passed to the `Subscriber` via the `onSubscribe` callback. -#### 4.Processor ([Code](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.2-RC2/api/src/main/java/org/reactivestreams/Processor.java)) +#### 4.Processor ([Code](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.3/api/src/main/java/org/reactivestreams/Processor.java)) ```java public interface Processor extends Subscriber, Publisher { diff --git a/RELEASE-NOTES.md b/RELEASE-NOTES.md index 840171f2..e91255f9 100644 --- a/RELEASE-NOTES.md +++ b/RELEASE-NOTES.md @@ -1,73 +1,176 @@ # Release notes for Reactive Streams ---- +# Version 1.0.3 released on 2019-08-23 + +## Announcement: + +We—the Reactive Streams community—are pleased to announce the immediate availability of `Reactive Streams 1.0.3`. This update to `Reactive Streams` brings the following improvements over `1.0.2`. + +## Highlights: + +- Specification + + Glossary term "External synchronization" has been [superseded](#Glossary) + + No breaking/semantical changes + + Rule [clarifications](#specification-clarifications-103-RC1) +- Interfaces + + No changes +- Technology Compatibility Kit (TCK) + + Improved [coverage](#tck-alterations-103-RC1) + + Improved JavaDoc +- Examples + + No changes +- Artifacts + + FlowAdapters artifact removed, FlowAdapters moved into the core jar ([#424](https://github.com/reactive-streams/reactive-streams-jvm/issues/424)) + +## Specification clarifications 1.0.3 + +## Glossary term "External synchronization" replaced by "Serial(ly)" + +**1.0.2:** Access coordination for thread safety purposes implemented outside of the constructs defined in this specification, using techniques such as, but not limited to, `atomics`, `monitors`, or `locks`. + +**1.0.3** In the context of a Signal, non-overlapping. In the context of the JVM, calls to methods on an object are serial if and only if there is a happens-before relationship between those calls (implying also that the calls do not overlap). When the calls are performed asynchronously, coordination to establish the happens-before relationship is to be implemented using techniques such as, but not limited to, atomics, monitors, or locks. + +## Publisher Rule 3 (Rule and Intent clarified) + +**1.0.2:** `onSubscribe`, `onNext`, `onError` and `onComplete` signaled to a `Subscriber` MUST be signaled in a thread-safe manner—and if performed by multiple threads—use external synchronization. + +*The intent of this rule is to make it clear that external synchronization must be employed if the Publisher intends to send signals from multiple/different threads.* + +**1.0.3:** `onSubscribe`, `onNext`, `onError` and `onComplete` signaled to a `Subscriber` MUST be signaled serially. + +*The intent of this rule is to permit the signalling of signals (including from multiple threads) if and only if a happens-before relation between each of the signals is established.* + +## Subscriber Rule 1 (Intent clarified) + +**1.0.2:** A `Subscriber` MUST signal demand via `Subscription.request(long n)` to receive `onNext` signals. + +*The intent of this rule is to establish that it is the responsibility of the Subscriber to signal when, and how many, elements it is able and willing to receive.* + +**1.0.3:** A `Subscriber` MUST signal demand via `Subscription.request(long n)` to receive `onNext` signals. + +*The intent of this rule is to establish that it is the responsibility of the Subscriber to decide when and how many elements it is able and willing to receive. To avoid signal reordering caused by reentrant Subscription methods, it is strongly RECOMMENDED for synchronous Subscriber implementations to invoke Subscription methods at the very end of any signal processing. It is RECOMMENDED that Subscribers request the upper limit of what they are able to process, as requesting only one element at a time results in an inherently inefficient "stop-and-wait" protocol.* + +## Subscriber Rule 5 (Intent clarified) + +**1.0.2:** A `Subscriber` MUST call `Subscription.cancel()` on the given `Subscription` after an `onSubscribe` signal if it already has an active `Subscription` + +*The intent of this rule is to prevent that two, or more, separate Publishers from thinking that they can interact with the same Subscriber. Enforcing this rule means that resource leaks are prevented since extra Subscriptions will be cancelled.* -# Version 1.0.2-RC2 released on 2017-12-07 +**1.0.3:** A `Subscriber` MUST call `Subscription.cancel()` on the given `Subscription` after an `onSubscribe` signal if it already has an active `Subscription` -- Addresses [some oversights](https://github.com/reactive-streams/reactive-streams-jvm/pull/419) in the Flow-TCK for Flow.Processor +*The intent of this rule is to prevent that two, or more, separate Publishers from trying to interact with the same Subscriber. Enforcing this rule means that resource leaks are prevented since extra Subscriptions will be cancelled. Failure to conform to this rule may lead to violations of Publisher rule 1, amongst others. Such violations can lead to hard-to-diagnose bugs.* -For more information on the 1.0.2 version, see below. +## Subscriber Rule 7 (Rule and Intent clarified) + +**1.0.2:** A `Subscriber` MUST ensure that all calls on its `Subscription` take place from the same thread or provide for respective external synchronization. + +*The intent of this rule is to establish that external synchronization must be added if a Subscriber will be using a Subscription concurrently by two or more threads.* + +**1.0.3:** A Subscriber MUST ensure that all calls on its Subscription's request and cancel methods are performed serially. + +*The intent of this rule is to permit the calling of the request and cancel methods (including from multiple threads) if and only if a happens-before relation between each of the calls is established.* + +## TCK alterations 1.0.3 + +- `PublisherVerification.optional_spec105_emptyStreamMustTerminateBySignallingOnComplete` fails if the publisher completes synchronously ([#422](https://github.com/reactive-streams/reactive-streams-jvm/issues/422)) +- IdentityFlowProcessorVerification throws NPE when `createFailedFlowPublisher` returns null ([#425](https://github.com/reactive-streams/reactive-streams-jvm/issues/425)) +- `required_spec208_mustBePreparedToReceiveOnNextSignalsAfterHavingCalledSubscriptionCancel` does not wait for request before invoking onNext ([#277](https://github.com/reactive-streams/reactive-streams-jvm/issues/277)) +- Subscriber whitebox verification tests demand ([#280](https://github.com/reactive-streams/reactive-streams-jvm/issues/280)) +- Incomplete documentation on stochastic tests in TCK ([#278](https://github.com/reactive-streams/reactive-streams-jvm/issues/278)) +- TCK performance ([#446](https://github.com/reactive-streams/reactive-streams-jvm/issues/446)) +- TCK: Receptacle#expectError timeout approach ([#451](https://github.com/reactive-streams/reactive-streams-jvm/issues/451)) + + +## Contributors + + Roland Kuhn [(@rkuhn)](https://github.com/rkuhn) + + Ben Christensen [(@benjchristensen)](https://github.com/benjchristensen) + + Viktor Klang [(@viktorklang)](https://github.com/viktorklang) + + Stephane Maldini [(@smaldini)](https://github.com/smaldini) + + Stanislav Savulchik [(@savulchik)](https://github.com/savulchik) + + Konrad Malawski [(@ktoso)](https://github.com/ktoso) + + Slim Ouertani [(@ouertani)](https://github.com/ouertani) + + Martynas Mickevičius [(@2m)](https://github.com/2m) + + Luke Daley [(@ldaley)](https://github.com/ldaley) + + Colin Godsey [(@colinrgodsey)](https://github.com/colinrgodsey) + + David Moten [(@davidmoten)](https://github.com/davidmoten) + + Brian Topping [(@briantopping)](https://github.com/briantopping) + + Rossen Stoyanchev [(@rstoyanchev)](https://github.com/rstoyanchev) + + Björn Hamels [(@BjornHamels)](https://github.com/BjornHamels) + + Jake Wharton [(@JakeWharton)](https://github.com/JakeWharton) + + Anthony Vanelverdinghe[(@anthonyvdotbe)](https://github.com/anthonyvdotbe) + + Kazuhiro Sera [(@seratch)](https://github.com/seratch) + + Dávid Karnok [(@akarnokd)](https://github.com/akarnokd) + + Evgeniy Getman [(@egetman)](https://github.com/egetman) + + Ángel Sanz [(@angelsanz)](https://github.com/angelsanz) + + shenghaiyang [(@shenghaiyang)](https://github.com/shenghaiyang) + + Kyle Thomson [(@kiiadi)](https://github.com/kiiadi) + + (new) James Roper [(@jroper)](https://github.com/jroper) + + (new) Oleh Dokuka [(@olegdokuka)](https://github.com/olegdokuka) + + (new) Scott Mitchell [(@Scottmitch)](https://github.com/Scottmitch) + + +--- -# Version 1.0.2-RC1 released on 2017-12-05 +# Version 1.0.2 released on 2017-12-19 ## Announcement: -We—the Reactive Streams community—are pleased to announce the immediate availability of `Reactive Streams 1.0.2-RC1`. This update to `Reactive Streams` brings the following improvements over `1.0.1`. +We—the Reactive Streams community—are pleased to announce the immediate availability of `Reactive Streams 1.0.2`. This update to `Reactive Streams` brings the following improvements over `1.0.1`. ## Highlights: - Specification + Glossary term added for `Thread-safe` + No breaking/semantical changes - + Rule [clarifications](#specification-clarifications-102-rc1) + + Rule [clarifications](#specification-clarifications-102) - Interfaces + No changes - Technology Compatibility Kit (TCK) - + Improved [coverage](#tck-alterations-102-rc1) - * Supports Publishers/Processors which do [coordinated emission](http://www.reactive-streams.org/reactive-streams-tck-1.0.2-RC1-javadoc/org/reactivestreams/tck/PublisherVerification.html#doesCoordinatedEmission--). + + Improved [coverage](#tck-alterations-102) + * Supports Publishers/Processors which do [coordinated emission](http://www.reactive-streams.org/reactive-streams-tck-1.0.2-javadoc/org/reactivestreams/tck/PublisherVerification.html#doesCoordinatedEmission--). + Improved JavaDoc - Examples - + New example [RangePublisher](http://www.reactive-streams.org/reactive-streams-examples-1.0.2-RC1-javadoc/org/reactivestreams/example/unicast/RangePublisher.html) + + New example [RangePublisher](http://www.reactive-streams.org/reactive-streams-examples-1.0.2-javadoc/org/reactivestreams/example/unicast/RangePublisher.html) - Artifacts + NEW! [Flow adapters](#flow-adapters) + NEW! [Flow TCK](#flow-tck) + Java 9 [Automatic-Module-Name](#automatic-module-name) added for all artifacts -## Specification clarifications 1.0.2-RC1 +## Specification clarifications 1.0.2 ## Subscriber Rule 2 **1.0.1:** The intent of this rule is that a Subscriber should not obstruct the progress of the Publisher from an execution point-of-view. In other words, the Subscriber should not starve the Publisher from CPU cycles. -**1.0.2-RC1:** The intent of this rule is that a Subscriber should not obstruct the progress of the Publisher from an execution point-of-view. In other words, the Subscriber should not starve the Publisher from receiving CPU cycles. +**1.0.2:** The intent of this rule is that a Subscriber should not obstruct the progress of the Publisher from an execution point-of-view. In other words, the Subscriber should not starve the Publisher from receiving CPU cycles. ## Subscriber Rule 8 **1.0.1:** The intent of this rule is to highlight that there may be a delay between calling `cancel` the Publisher seeing that. -**1.0.2-RC1** The intent of this rule is to highlight that there may be a delay between calling `cancel` and the Publisher observing that cancellation. +**1.0.2** The intent of this rule is to highlight that there may be a delay between calling `cancel` and the Publisher observing that cancellation. ## Flow adapters -An adapter library has been created to convert `org.reactivestreams` to `java.util.concurrent.Flow` and vice versa. Read more about it [here](http://www.reactive-streams.org/reactive-streams-flow-adapters-1.0.2-RC1-javadoc) +An adapter library has been created to convert `org.reactivestreams` to `java.util.concurrent.Flow` and vice versa. Read more about it [here](http://www.reactive-streams.org/reactive-streams-flow-adapters-1.0.2-javadoc) ~~~xml org.reactivestreams reactive-streams-flow-adapters - 1.0.2-RC1 + 1.0.2 ~~~ ## Flow TCK -A TCK artifact has been created to allow for direct TCK verification of `java.util.concurrent.Flow` implementations. Read more about it [here](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.2-RC1/tck-flow/README.md) +A TCK artifact has been created to allow for direct TCK verification of `java.util.concurrent.Flow` implementations. Read more about it [here](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.2/tck-flow/README.md) ~~~xml org.reactivestreams reactive-streams-tck-flow - 1.0.2-RC1 + 1.0.2 ~~~ @@ -79,7 +182,7 @@ A TCK artifact has been created to allow for direct TCK verification of `java.ut * `org.reactivestreams:reactive-streams-flow-adapters` => `org.reactivestreams.flowadapters` * `org.reactivestreams:reactive-streams-tck-flow` => `org.reactivestreams.tckflow` -## TCK alterations 1.0.2-RC1 +## TCK alterations 1.0.2 - Added support for Publisher verification of Publishers who do coordinated emission, i.e. where elements only are emitted after all current Subscribers have signalled demand. ([#284](https://github.com/reactive-streams/reactive-streams-jvm/issues/284)) - The `SubscriberWhiteboxVerification` has been given more user friendly error messages in the case where the user forgets to call `registerOnSubscribe`. (#416)[https://github.com/reactive-streams/reactive-streams-jvm/pull/416] diff --git a/api/build.gradle b/api/build.gradle index 6529a2bf..f8fd005f 100644 --- a/api/build.gradle +++ b/api/build.gradle @@ -1,4 +1,22 @@ description = "reactive-streams" +def jdkFlow = false +try { + Class.forName("java.util.concurrent.Flow") + jdkFlow = true +} catch (ClassNotFoundException cnfe) { + +} + +sourceSets { + main { + java { + if (jdkFlow) + srcDirs = ['src/main/java', 'src/main/java9'] + else + srcDirs = ['src/main/java'] + } + } +} jar { manifest { diff --git a/api/src/main/java/org/reactivestreams/Subscription.java b/api/src/main/java/org/reactivestreams/Subscription.java index c18d0fc9..167f6a87 100644 --- a/api/src/main/java/org/reactivestreams/Subscription.java +++ b/api/src/main/java/org/reactivestreams/Subscription.java @@ -23,8 +23,8 @@ public interface Subscription { /** * No events will be sent by a {@link Publisher} until demand is signaled via this method. *

- * It can be called however often and whenever needed—but the outstanding cumulative demand must never exceed Long.MAX_VALUE. - * An outstanding cumulative demand of Long.MAX_VALUE may be treated by the {@link Publisher} as "effectively unbounded". + * It can be called however often and whenever needed—but if the outstanding cumulative demand ever becomes Long.MAX_VALUE or more, + * it may be treated by the {@link Publisher} as "effectively unbounded". *

* Whatever has been requested can be sent by the {@link Publisher} so only signal demand for what can be safely handled. *

diff --git a/flow-adapters/src/main/java/org/reactivestreams/FlowAdapters.java b/api/src/main/java9/org/reactivestreams/FlowAdapters.java similarity index 74% rename from flow-adapters/src/main/java/org/reactivestreams/FlowAdapters.java rename to api/src/main/java9/org/reactivestreams/FlowAdapters.java index 4417e4c5..96cbff5b 100644 --- a/flow-adapters/src/main/java/org/reactivestreams/FlowAdapters.java +++ b/api/src/main/java9/org/reactivestreams/FlowAdapters.java @@ -12,6 +12,7 @@ package org.reactivestreams; import java.util.concurrent.Flow; +import static java.util.Objects.requireNonNull; /** * Bridge between Reactive Streams API and the Java 9 {@link java.util.concurrent.Flow} API. @@ -31,16 +32,16 @@ private FlowAdapters() { @SuppressWarnings("unchecked") public static org.reactivestreams.Publisher toPublisher( Flow.Publisher flowPublisher) { - if (flowPublisher == null) { - throw new NullPointerException("flowPublisher"); - } + requireNonNull(flowPublisher, "flowPublisher"); + final org.reactivestreams.Publisher publisher; if (flowPublisher instanceof FlowPublisherFromReactive) { - return (org.reactivestreams.Publisher)(((FlowPublisherFromReactive)flowPublisher).reactiveStreams); - } - if (flowPublisher instanceof org.reactivestreams.Publisher) { - return (org.reactivestreams.Publisher)flowPublisher; + publisher = (org.reactivestreams.Publisher)(((FlowPublisherFromReactive)flowPublisher).reactiveStreams); + } else if (flowPublisher instanceof org.reactivestreams.Publisher) { + publisher = (org.reactivestreams.Publisher)flowPublisher; + } else { + publisher = new ReactivePublisherFromFlow(flowPublisher); } - return new ReactivePublisherFromFlow(flowPublisher); + return publisher; } /** @@ -53,16 +54,16 @@ public static org.reactivestreams.Publisher toPublisher( public static Flow.Publisher toFlowPublisher( org.reactivestreams.Publisher reactiveStreamsPublisher ) { - if (reactiveStreamsPublisher == null) { - throw new NullPointerException("reactiveStreamsPublisher"); - } + requireNonNull(reactiveStreamsPublisher, "reactiveStreamsPublisher"); + final Flow.Publisher flowPublisher; if (reactiveStreamsPublisher instanceof ReactivePublisherFromFlow) { - return (Flow.Publisher)(((ReactivePublisherFromFlow)reactiveStreamsPublisher).flow); - } - if (reactiveStreamsPublisher instanceof Flow.Publisher) { - return (Flow.Publisher)reactiveStreamsPublisher; + flowPublisher = (Flow.Publisher)(((ReactivePublisherFromFlow)reactiveStreamsPublisher).flow); + } else if (reactiveStreamsPublisher instanceof Flow.Publisher) { + flowPublisher = (Flow.Publisher)reactiveStreamsPublisher; + } else { + flowPublisher = new FlowPublisherFromReactive(reactiveStreamsPublisher); } - return new FlowPublisherFromReactive(reactiveStreamsPublisher); + return flowPublisher; } /** @@ -76,16 +77,16 @@ public static Flow.Publisher toFlowPublisher( public static org.reactivestreams.Processor toProcessor( Flow.Processor flowProcessor ) { - if (flowProcessor == null) { - throw new NullPointerException("flowProcessor"); - } + requireNonNull(flowProcessor, "flowProcessor"); + final org.reactivestreams.Processor processor; if (flowProcessor instanceof FlowToReactiveProcessor) { - return (org.reactivestreams.Processor)(((FlowToReactiveProcessor)flowProcessor).reactiveStreams); - } - if (flowProcessor instanceof org.reactivestreams.Processor) { - return (org.reactivestreams.Processor)flowProcessor; + processor = (org.reactivestreams.Processor)(((FlowToReactiveProcessor)flowProcessor).reactiveStreams); + } else if (flowProcessor instanceof org.reactivestreams.Processor) { + processor = (org.reactivestreams.Processor)flowProcessor; + } else { + processor = new ReactiveToFlowProcessor(flowProcessor); } - return new ReactiveToFlowProcessor(flowProcessor); + return processor; } /** @@ -99,16 +100,16 @@ public static org.reactivestreams.Processor toProcessor( public static Flow.Processor toFlowProcessor( org.reactivestreams.Processor reactiveStreamsProcessor ) { - if (reactiveStreamsProcessor == null) { - throw new NullPointerException("reactiveStreamsProcessor"); - } + requireNonNull(reactiveStreamsProcessor, "reactiveStreamsProcessor"); + final Flow.Processor flowProcessor; if (reactiveStreamsProcessor instanceof ReactiveToFlowProcessor) { - return (Flow.Processor)(((ReactiveToFlowProcessor)reactiveStreamsProcessor).flow); + flowProcessor = (Flow.Processor)(((ReactiveToFlowProcessor)reactiveStreamsProcessor).flow); + } else if (reactiveStreamsProcessor instanceof Flow.Processor) { + flowProcessor = (Flow.Processor)reactiveStreamsProcessor; + } else { + flowProcessor = new FlowToReactiveProcessor(reactiveStreamsProcessor); } - if (reactiveStreamsProcessor instanceof Flow.Processor) { - return (Flow.Processor)reactiveStreamsProcessor; - } - return new FlowToReactiveProcessor(reactiveStreamsProcessor); + return flowProcessor; } /** @@ -119,16 +120,16 @@ public static Flow.Processor toFlowProcessor( */ @SuppressWarnings("unchecked") public static Flow.Subscriber toFlowSubscriber(org.reactivestreams.Subscriber reactiveStreamsSubscriber) { - if (reactiveStreamsSubscriber == null) { - throw new NullPointerException("reactiveStreamsSubscriber"); - } + requireNonNull(reactiveStreamsSubscriber, "reactiveStreamsSubscriber"); + final Flow.Subscriber flowSubscriber; if (reactiveStreamsSubscriber instanceof ReactiveToFlowSubscriber) { - return (Flow.Subscriber)((ReactiveToFlowSubscriber)reactiveStreamsSubscriber).flow; + flowSubscriber = (Flow.Subscriber)((ReactiveToFlowSubscriber)reactiveStreamsSubscriber).flow; + } else if (reactiveStreamsSubscriber instanceof Flow.Subscriber) { + flowSubscriber = (Flow.Subscriber)reactiveStreamsSubscriber; + } else { + flowSubscriber = new FlowToReactiveSubscriber(reactiveStreamsSubscriber); } - if (reactiveStreamsSubscriber instanceof Flow.Subscriber) { - return (Flow.Subscriber)reactiveStreamsSubscriber; - } - return new FlowToReactiveSubscriber(reactiveStreamsSubscriber); + return flowSubscriber; } /** @@ -139,16 +140,16 @@ public static Flow.Subscriber toFlowSubscriber(org.reactivestreams.Subscr */ @SuppressWarnings("unchecked") public static org.reactivestreams.Subscriber toSubscriber(Flow.Subscriber flowSubscriber) { - if (flowSubscriber == null) { - throw new NullPointerException("flowSubscriber"); - } + requireNonNull(flowSubscriber, "flowSubscriber"); + final org.reactivestreams.Subscriber subscriber; if (flowSubscriber instanceof FlowToReactiveSubscriber) { - return (org.reactivestreams.Subscriber)((FlowToReactiveSubscriber)flowSubscriber).reactiveStreams; + subscriber = (org.reactivestreams.Subscriber)((FlowToReactiveSubscriber)flowSubscriber).reactiveStreams; + } else if (flowSubscriber instanceof org.reactivestreams.Subscriber) { + subscriber = (org.reactivestreams.Subscriber)flowSubscriber; + } else { + subscriber = new ReactiveToFlowSubscriber(flowSubscriber); } - if (flowSubscriber instanceof org.reactivestreams.Subscriber) { - return (org.reactivestreams.Subscriber)flowSubscriber; - } - return new ReactiveToFlowSubscriber(flowSubscriber); + return subscriber; } /** @@ -200,8 +201,7 @@ public void cancel() { * Wraps a Reactive Streams Subscriber and forwards methods of the Flow Subscriber to it. * @param the element type */ - static final class FlowToReactiveSubscriber - implements Flow.Subscriber { + static final class FlowToReactiveSubscriber implements Flow.Subscriber { final org.reactivestreams.Subscriber reactiveStreams; public FlowToReactiveSubscriber(org.reactivestreams.Subscriber reactive) { @@ -231,11 +231,10 @@ public void onComplete() { } /** - * Wraps a Reactive Streams Subscriber and forwards methods of the Flow Subscriber to it. + * Wraps a Flow Subscriber and forwards methods of the Reactive Streams Subscriber to it. * @param the element type */ - static final class ReactiveToFlowSubscriber - implements org.reactivestreams.Subscriber { + static final class ReactiveToFlowSubscriber implements org.reactivestreams.Subscriber { final Flow.Subscriber flow; public ReactiveToFlowSubscriber(Flow.Subscriber flow) { @@ -269,8 +268,7 @@ public void onComplete() { * @param the input type * @param the output type */ - static final class ReactiveToFlowProcessor - implements org.reactivestreams.Processor { + static final class ReactiveToFlowProcessor implements org.reactivestreams.Processor { final Flow.Processor flow; public ReactiveToFlowProcessor(Flow.Processor flow) { @@ -308,8 +306,7 @@ public void subscribe(org.reactivestreams.Subscriber s) { * @param the input type * @param the output type */ - static final class FlowToReactiveProcessor - implements Flow.Processor { + static final class FlowToReactiveProcessor implements Flow.Processor { final org.reactivestreams.Processor reactiveStreams; public FlowToReactiveProcessor(org.reactivestreams.Processor reactive) { @@ -347,7 +344,6 @@ public void subscribe(Flow.Subscriber s) { * @param the element type */ static final class ReactivePublisherFromFlow implements org.reactivestreams.Publisher { - final Flow.Publisher flow; public ReactivePublisherFromFlow(Flow.Publisher flowPublisher) { diff --git a/build.gradle b/build.gradle index 6f562383..9454d59b 100644 --- a/build.gradle +++ b/build.gradle @@ -3,7 +3,7 @@ subprojects { apply plugin: "osgi" group = "org.reactivestreams" - version = "1.0.2-RC2" + version = "1.0.3" sourceCompatibility = 1.6 targetCompatibility = 1.6 @@ -42,17 +42,17 @@ subprojects { instructionReplace "Bundle-Vendor", "Reactive Streams SIG" instructionReplace "Bundle-Description", "Reactive Streams API" instructionReplace "Bundle-DocURL", "http://reactive-streams.org" - instructionReplace "Bundle-Version", "1.0.2-RC2" + instructionReplace "Bundle-Version", "1.0.3" } } if (name in ["reactive-streams", "reactive-streams-tck", "reactive-streams-tck-flow", - "reactive-streams-examples", - "reactive-streams-flow-adapters"]) { + "reactive-streams-examples"]) { apply plugin: "maven" apply plugin: "signing" + apply plugin: "maven-publish" signing { sign configurations.archives @@ -68,6 +68,17 @@ subprojects { from javadoc } + publishing { + publications { + mavenJava(MavenPublication) { + from components.java + } + } + repositories { + mavenLocal() + } + } + artifacts { archives sourcesJar, javadocJar } diff --git a/flow-adapters/.gitignore b/flow-adapters/.gitignore deleted file mode 100644 index 7447f89a..00000000 --- a/flow-adapters/.gitignore +++ /dev/null @@ -1 +0,0 @@ -/bin \ No newline at end of file diff --git a/flow-adapters/build.gradle b/flow-adapters/build.gradle deleted file mode 100644 index 8a7f1a96..00000000 --- a/flow-adapters/build.gradle +++ /dev/null @@ -1,20 +0,0 @@ -description = 'reactive-streams-flow-adapters' - -dependencies { - compile project(':reactive-streams') - - testCompile project(':reactive-streams-tck') - testCompile group: 'org.testng', name: 'testng', version: '5.14.10' -} - -jar { - manifest { - attributes('Automatic-Module-Name': 'org.reactivestreams.flowadapters') - } -} - -test.useTestNG() - -javadoc { - options.links("https://docs.oracle.com/javase/9/docs/api") -} \ No newline at end of file diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties index 12556c82..1d5b29fb 100644 --- a/gradle/wrapper/gradle-wrapper.properties +++ b/gradle/wrapper/gradle-wrapper.properties @@ -1,5 +1,5 @@ -distributionUrl=http\://services.gradle.org/distributions/gradle-4.3-bin.zip distributionBase=GRADLE_USER_HOME distributionPath=wrapper/dists -zipStorePath=wrapper/dists zipStoreBase=GRADLE_USER_HOME +zipStorePath=wrapper/dists +distributionUrl=https\://services.gradle.org/distributions/gradle-4.10.3-bin.zip diff --git a/settings.gradle b/settings.gradle index a1cbab78..866ecf93 100644 --- a/settings.gradle +++ b/settings.gradle @@ -11,12 +11,12 @@ try { Class.forName("java.util.concurrent.Flow") jdkFlow = true println(ANSI_GREEN + " INFO: ------------------ JDK9 classes detected ---------------------------------" + ANSI_RESET) - println(ANSI_GREEN + " INFO: Java 9 Flow API found; Including [flow-adapters, tck-flow] in build. " + ANSI_RESET) + println(ANSI_GREEN + " INFO: Java 9 Flow API found; Including [tck-flow] & FlowAdapters in build. " + ANSI_RESET) println(ANSI_GREEN + " INFO: --------------------------------------------------------------------------" + ANSI_RESET) } catch (Throwable ex) { // Flow API not available println(ANSI_RED + "WARNING: -------------------- JDK9 classes NOT detected -----------------------------" + ANSI_RESET) - println(ANSI_RED + "WARNING: Java 9 Flow API not found; Not including [flow-adapters, tck-flow] in build." + ANSI_RESET) + println(ANSI_RED + "WARNING: Java 9 Flow API not found; Not including [tck-flow] & FlowAdapters in build." + ANSI_RESET) println(ANSI_RED + "WARNING: In order to execute the complete test-suite run the build using JDK9+. " + ANSI_RESET) println(ANSI_RED + "WARNING: ----------------------------------------------------------------------------" + ANSI_RESET) } @@ -26,7 +26,6 @@ include ':reactive-streams-tck' include ':reactive-streams-examples' if (jdkFlow) { - include ':reactive-streams-flow-adapters' include ':reactive-streams-tck-flow' } @@ -34,6 +33,5 @@ project(':reactive-streams').projectDir = "$rootDir/api" as File project(':reactive-streams-tck').projectDir = "$rootDir/tck" as File project(':reactive-streams-examples').projectDir = "$rootDir/examples" as File if (jdkFlow) { - project(':reactive-streams-flow-adapters').projectDir = "$rootDir/flow-adapters" as File project(':reactive-streams-tck-flow').projectDir = "$rootDir/tck-flow" as File } diff --git a/tck-flow/README.md b/tck-flow/README.md index e449f95f..a007541d 100644 --- a/tck-flow/README.md +++ b/tck-flow/README.md @@ -27,7 +27,7 @@ The TCK is provided as binary artifact on [Maven Central](http://search.maven.or org.reactivestreams reactive-streams-tck-flow - 1.0.2-RC2 + 1.0.3 test ``` @@ -209,18 +209,24 @@ within the TCK which await for something to happen. The other timeout is `publis [Rule 3.13](https://github.com/reactive-streams/reactive-streams-jvm#3.13) which defines that `Subscriber` references MUST be dropped by the Publisher. -Note that the TCK differenciates between timeouts for "waiting for a signal" (``defaultTimeoutMillis``), -and "asserting no signals happen during a given amount of time" (``envDefaultNoSignalsTimeoutMillis``). -While the latter defaults to the prior, it may be useful to tweak them independently when running on continious -integration servers (for example, keeping the no-signals timeout significantly lower). +Note that the TCK differentiates between timeouts for "waiting for a signal" +(`defaultTimeoutMillis`), and "asserting no signals happen during a given amount of time" +(`defaultNoSignalsTimeoutMillis`). While the latter defaults to the prior, it may be useful to tweak +them independently when running on continuous integration servers (for example, keeping the +no-signals timeout significantly lower). Another configuration option is the "poll timeout" which is +used whenever an operation has to poll for a `defaultTimeoutMillis` for a signal to appear (most +often errors), it can then poll and check using the `defaultPollTimeoutMillis`, for the expected +error, rather than blocking for the full default timeout. -In order to configure these timeouts (for example when running on a slow continious integtation machine), you can either: +In order to configure these timeouts (for example when running on a slow continuous integration +machine), you can either: **Use env variables** to set these timeouts, in which case the you can do: ```bash export DEFAULT_TIMEOUT_MILLIS=100 export DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS=100 +export DEFAULT_POLL_TIMEOUT_MILLIS=20 export PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS=300 ``` @@ -231,10 +237,11 @@ public class RangePublisherTest extends FlowPublisherVerification { public static final long DEFAULT_TIMEOUT_MILLIS = 100L; public static final long DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS = DEFAULT_TIMEOUT_MILLIS; - public static final long PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS = 500L; + public static final long DEFAULT_POLL_TIMEOUT_MILLIS = 20L; + public static final long PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS = 300L; public RangePublisherTest() { - super(new TestEnvironment(DEFAULT_TIMEOUT_MILLIS, DEFAULT_TIMEOUT_MILLIS), PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS); + super(new TestEnvironment(DEFAULT_TIMEOUT_MILLIS, DEFAULT_TIMEOUT_MILLIS, DEFAULT_POLL_TIMEOUT_MILLIS), PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS); } // ... diff --git a/tck-flow/build.gradle b/tck-flow/build.gradle index 2ad8aa94..bf661453 100644 --- a/tck-flow/build.gradle +++ b/tck-flow/build.gradle @@ -2,7 +2,6 @@ description = 'reactive-streams-tck-flow' dependencies { compile group: 'org.testng', name: 'testng', version:'5.14.10' compile project(':reactive-streams-tck') - compile project(':reactive-streams-flow-adapters') } jar { diff --git a/tck-flow/src/main/java/org/reactivestreams/tck/flow/IdentityFlowProcessorVerification.java b/tck-flow/src/main/java/org/reactivestreams/tck/flow/IdentityFlowProcessorVerification.java index f4479e71..e38ed780 100644 --- a/tck-flow/src/main/java/org/reactivestreams/tck/flow/IdentityFlowProcessorVerification.java +++ b/tck-flow/src/main/java/org/reactivestreams/tck/flow/IdentityFlowProcessorVerification.java @@ -60,7 +60,9 @@ public final Processor createIdentityProcessor(int bufferSize) { @Override public final Publisher createFailedPublisher() { - return FlowAdapters.toPublisher(createFailedFlowPublisher()); + Flow.Publisher failed = createFailedFlowPublisher(); + if (failed == null) return null; // because `null` means "SKIP" in createFailedPublisher + else return FlowAdapters.toPublisher(failed); } } diff --git a/flow-adapters/src/test/java/org/reactivestreams/FlowAdaptersTest.java b/tck-flow/src/test/java/org/reactivestreams/FlowAdaptersTest.java similarity index 100% rename from flow-adapters/src/test/java/org/reactivestreams/FlowAdaptersTest.java rename to tck-flow/src/test/java/org/reactivestreams/FlowAdaptersTest.java diff --git a/flow-adapters/src/test/java/org/reactivestreams/MulticastPublisher.java b/tck-flow/src/test/java/org/reactivestreams/MulticastPublisher.java similarity index 100% rename from flow-adapters/src/test/java/org/reactivestreams/MulticastPublisher.java rename to tck-flow/src/test/java/org/reactivestreams/MulticastPublisher.java diff --git a/flow-adapters/src/test/java/org/reactivestreams/SubmissionPublisherTckTest.java b/tck-flow/src/test/java/org/reactivestreams/SubmissionPublisherTckTest.java similarity index 100% rename from flow-adapters/src/test/java/org/reactivestreams/SubmissionPublisherTckTest.java rename to tck-flow/src/test/java/org/reactivestreams/SubmissionPublisherTckTest.java diff --git a/flow-adapters/src/test/java/org/reactivestreams/TestEitherConsumer.java b/tck-flow/src/test/java/org/reactivestreams/TestEitherConsumer.java similarity index 100% rename from flow-adapters/src/test/java/org/reactivestreams/TestEitherConsumer.java rename to tck-flow/src/test/java/org/reactivestreams/TestEitherConsumer.java diff --git a/tck/README.md b/tck/README.md index 8c4a318c..93dfe620 100644 --- a/tck/README.md +++ b/tck/README.md @@ -27,7 +27,7 @@ The TCK is provided as binary artifact on [Maven Central](http://search.maven.or org.reactivestreams reactive-streams-tck - 1.0.2-RC2 + 1.0.3 test ``` @@ -209,18 +209,24 @@ within the TCK which await for something to happen. The other timeout is `publis [Rule 3.13](https://github.com/reactive-streams/reactive-streams-jvm#3.13) which defines that `Subscriber` references MUST be dropped by the Publisher. -Note that the TCK differenciates between timeouts for "waiting for a signal" (``defaultTimeoutMillis``), -and "asserting no signals happen during a given amount of time" (``envDefaultNoSignalsTimeoutMillis``). -While the latter defaults to the prior, it may be useful to tweak them independently when running on continious -integration servers (for example, keeping the no-signals timeout significantly lower). +Note that the TCK differentiates between timeouts for "waiting for a signal" +(`defaultTimeoutMillis`), and "asserting no signals happen during a given amount of time" +(`defaultNoSignalsTimeoutMillis`). While the latter defaults to the prior, it may be useful to tweak +them independently when running on continuous integration servers (for example, keeping the +no-signals timeout significantly lower). Another configuration option is the "poll timeout" which is +used whenever an operation has to poll for a `defaultTimeoutMillis` for a signal to appear (most +often errors), it can then poll and check using the `defaultPollTimeoutMillis`, for the expected +error, rather than blocking for the full default timeout. -In order to configure these timeouts (for example when running on a slow continious integtation machine), you can either: +In order to configure these timeouts (for example when running on a slow continuous integration +machine), you can either: **Use env variables** to set these timeouts, in which case the you can do: ```bash export DEFAULT_TIMEOUT_MILLIS=100 export DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS=100 +export DEFAULT_POLL_TIMEOUT_MILLIS=20 export PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS=300 ``` @@ -231,10 +237,11 @@ public class RangePublisherTest extends PublisherVerification { public static final long DEFAULT_TIMEOUT_MILLIS = 100L; public static final long DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS = DEFAULT_TIMEOUT_MILLIS; - public static final long PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS = 500L; + public static final long DEFAULT_POLL_TIMEOUT_MILLIS = 20L; + public static final long PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS = 300L; public RangePublisherTest() { - super(new TestEnvironment(DEFAULT_TIMEOUT_MILLIS, DEFAULT_TIMEOUT_MILLIS), PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS); + super(new TestEnvironment(DEFAULT_TIMEOUT_MILLIS, DEFAULT_TIMEOUT_MILLIS, DEFAULT_POLL_TIMEOUT_MILLIS), PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS); } // ... diff --git a/tck/src/main/java/org/reactivestreams/tck/IdentityProcessorVerification.java b/tck/src/main/java/org/reactivestreams/tck/IdentityProcessorVerification.java index 361e0821..53b74d05 100644 --- a/tck/src/main/java/org/reactivestreams/tck/IdentityProcessorVerification.java +++ b/tck/src/main/java/org/reactivestreams/tck/IdentityProcessorVerification.java @@ -172,7 +172,9 @@ public long boundedDepthOfOnNextAndRequestRecursion() { /** * Override and return {@code true} in order to skip executing tests marked as {@code Stochastic}. - * Such tests MAY sometimes fail even though the impl + * Stochastic in this case means that the Rule is impossible or infeasible to deterministically verify— + * usually this means that this test case can yield false positives ("be green") even if for some case, + * the given implementation may violate the tested behaviour. */ public boolean skipStochasticTests() { return false; @@ -502,7 +504,9 @@ public Subscriber createSubscriber(final SubscriberWhiteboxVerification.White @Override public void onSubscribe(final Subscription subscription) { - env.debug(String.format("whiteboxSubscriber::onSubscribe(%s)", subscription)); + if (env.debugEnabled()) { + env.debug(String.format("whiteboxSubscriber::onSubscribe(%s)", subscription)); + } if (subs.isCompleted()) subscription.cancel(); // the Probe must also pass subscriber verification probe.registerOnSubscribe(new SubscriberWhiteboxVerification.SubscriberPuppet() { @@ -521,19 +525,25 @@ public void signalCancel() { @Override public void onNext(T element) { - env.debug(String.format("whiteboxSubscriber::onNext(%s)", element)); + if (env.debugEnabled()) { + env.debug(String.format("whiteboxSubscriber::onNext(%s)", element)); + } probe.registerOnNext(element); } @Override public void onComplete() { - env.debug("whiteboxSubscriber::onComplete()"); + if (env.debugEnabled()) { + env.debug("whiteboxSubscriber::onComplete()"); + } probe.registerOnComplete(); } @Override public void onError(Throwable cause) { - env.debug(String.format("whiteboxSubscriber::onError(%s)", cause)); + if (env.debugEnabled()) { + env.debug(String.format("whiteboxSubscriber::onError(%s)", cause)); + } probe.registerOnError(cause); } }); diff --git a/tck/src/main/java/org/reactivestreams/tck/PublisherVerification.java b/tck/src/main/java/org/reactivestreams/tck/PublisherVerification.java index 54badd61..56af3ddb 100644 --- a/tck/src/main/java/org/reactivestreams/tck/PublisherVerification.java +++ b/tck/src/main/java/org/reactivestreams/tck/PublisherVerification.java @@ -772,7 +772,9 @@ public void onNext(T element) { signalsReceived += 1; stackDepthCounter.set(stackDepthCounter.get() + 1); - env.debug(String.format("%s(recursion depth: %d)::onNext(%s)", this, stackDepthCounter.get(), element)); + if (env.debugEnabled()) { + env.debug(String.format("%s(recursion depth: %d)::onNext(%s)", this, stackDepthCounter.get(), element)); + } final long callsUntilNow = stackDepthCounter.get(); if (callsUntilNow > boundedDepthOfOnNextAndRequestRecursion()) { @@ -1082,7 +1084,9 @@ public void required_spec317_mustNotSignalOnErrorWhenPendingAboveLongMaxValue() @Override public void onNext(T element) { - env.debug(String.format("%s::onNext(%s)", this, element)); + if (env.debugEnabled()) { + env.debug(String.format("%s::onNext(%s)", this, element)); + } if (subscription.isCompleted()) { if (callsCounter > 0) { subscription.value().request(Long.MAX_VALUE - 1); diff --git a/tck/src/main/java/org/reactivestreams/tck/SubscriberWhiteboxVerification.java b/tck/src/main/java/org/reactivestreams/tck/SubscriberWhiteboxVerification.java index 57970c46..3b14691f 100644 --- a/tck/src/main/java/org/reactivestreams/tck/SubscriberWhiteboxVerification.java +++ b/tck/src/main/java/org/reactivestreams/tck/SubscriberWhiteboxVerification.java @@ -266,7 +266,9 @@ public void required_spec208_mustBePreparedToReceiveOnNextSignalsAfterHavingCall @Override public void run(WhiteboxTestStage stage) throws InterruptedException { stage.puppet().triggerRequest(1); + stage.expectRequest(); stage.puppet().signalCancel(); + stage.expectCancelling(); stage.signalNext(); stage.puppet().triggerRequest(1); @@ -437,7 +439,12 @@ public void required_spec308_requestMustRegisterGivenNumberElementsToBeProduced( @Override public void run(WhiteboxTestStage stage) throws InterruptedException { stage.puppet().triggerRequest(2); + long requestedElements = stage.expectRequest(); stage.probe.expectNext(stage.signalNext()); + // Some subscribers may only request one element at a time. + if (requestedElements < 2) { + stage.expectRequest(); + } stage.probe.expectNext(stage.signalNext()); stage.probe.expectNone(); @@ -789,13 +796,34 @@ public interface SubscriberProbe { * make implement these calls. */ public interface SubscriberPuppet { + /** - * Trigger {@code request(elements)} on your {@link Subscriber} + * Ensure that at least {@code elements} are eventually requested by your {@link Subscriber}, if it hasn't already + * requested that many elements. + *

+ * This does not necessarily have to correlate 1:1 with a {@code Subscription.request(elements)} call, but the sum + * of the elements requested by your {@code Subscriber} must eventually be at least the the sum of the elements + * triggered to be requested by all the invocations of this method. + *

+ * Additionally, subscribers are permitted to delay requesting elements until previous requests for elements have + * been fulfilled. For example, a subscriber that only requests one element at a time may fulfill the request made + * by this method by requesting one element {@code elements} times, waiting for each element to arrive before the + * next request is made. + *

+ * Before sending any element to the subscriber, the TCK must wait for the subscriber to request that element, and + * must be prepared for the subscriber to only request one element at a time, it is not enough for the TCK to + * simply invoke this method before sending elements. + *

+ * An invocation of {@link #signalCancel()} may be coalesced into any elements that have not yet been requested, + * such that only a cancel signal is emitted. */ void triggerRequest(long elements); /** - * Trigger {@code cancel()} on your {@link Subscriber} + * Trigger {@code cancel()} on your {@link Subscriber}. + *

+ * An invocation of this method may be coalesced into any outstanding requests, as requested by + * {@link #triggerRequest(long)}, such that only a cancel signal is emitted. */ void signalCancel(); } diff --git a/tck/src/main/java/org/reactivestreams/tck/TestEnvironment.java b/tck/src/main/java/org/reactivestreams/tck/TestEnvironment.java index 8fdad9fc..f3e435cf 100644 --- a/tck/src/main/java/org/reactivestreams/tck/TestEnvironment.java +++ b/tck/src/main/java/org/reactivestreams/tck/TestEnvironment.java @@ -14,8 +14,8 @@ import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; -import org.reactivestreams.tck.flow.support.SubscriberBufferOverflowException; import org.reactivestreams.tck.flow.support.Optional; +import org.reactivestreams.tck.flow.support.SubscriberBufferOverflowException; import java.util.Collections; import java.util.LinkedList; @@ -24,7 +24,10 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.NANOSECONDS; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; @@ -35,8 +38,10 @@ public class TestEnvironment { private static final long DEFAULT_TIMEOUT_MILLIS = 100; private static final String DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS_ENV = "DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS"; + private static final String DEFAULT_POLL_TIMEOUT_MILLIS_ENV = "DEFAULT_POLL_TIMEOUT_MILLIS_ENV"; private final long defaultTimeoutMillis; + private final long defaultPollTimeoutMillis; private final long defaultNoSignalsTimeoutMillis; private final boolean printlnDebug; @@ -49,14 +54,46 @@ public class TestEnvironment { * run the tests. * @param defaultTimeoutMillis default timeout to be used in all expect* methods * @param defaultNoSignalsTimeoutMillis default timeout to be used when no further signals are expected anymore + * @param defaultPollTimeoutMillis default amount of time to poll for events if {@code defaultTimeoutMillis} isn't + * preempted by an asynchronous event. * @param printlnDebug if true, signals such as OnNext / Request / OnComplete etc will be printed to standard output, */ - public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMillis, boolean printlnDebug) { + public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMillis, long defaultPollTimeoutMillis, + boolean printlnDebug) { this.defaultTimeoutMillis = defaultTimeoutMillis; + this.defaultPollTimeoutMillis = defaultPollTimeoutMillis; this.defaultNoSignalsTimeoutMillis = defaultNoSignalsTimeoutMillis; this.printlnDebug = printlnDebug; } + /** + * Tests must specify the timeout for expected outcome of asynchronous + * interactions. Longer timeout does not invalidate the correctness of + * the implementation, but can in some cases result in longer time to + * run the tests. + * @param defaultTimeoutMillis default timeout to be used in all expect* methods + * @param defaultNoSignalsTimeoutMillis default timeout to be used when no further signals are expected anymore + * @param printlnDebug if true, signals such as OnNext / Request / OnComplete etc will be printed to standard output, + */ + public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMillis, boolean printlnDebug) { + this(defaultTimeoutMillis, defaultNoSignalsTimeoutMillis, defaultTimeoutMillis, printlnDebug); + } + + /** + * Tests must specify the timeout for expected outcome of asynchronous + * interactions. Longer timeout does not invalidate the correctness of + * the implementation, but can in some cases result in longer time to + * run the tests. + * + * @param defaultTimeoutMillis default timeout to be used in all expect* methods + * @param defaultNoSignalsTimeoutMillis default timeout to be used when no further signals are expected anymore + * @param defaultPollTimeoutMillis default amount of time to poll for events if {@code defaultTimeoutMillis} isn't + * preempted by an asynchronous event. + */ + public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMillis, long defaultPollTimeoutMillis) { + this(defaultTimeoutMillis, defaultNoSignalsTimeoutMillis, defaultPollTimeoutMillis, false); + } + /** * Tests must specify the timeout for expected outcome of asynchronous * interactions. Longer timeout does not invalidate the correctness of @@ -67,7 +104,7 @@ public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMi * @param defaultNoSignalsTimeoutMillis default timeout to be used when no further signals are expected anymore */ public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMillis) { - this(defaultTimeoutMillis, defaultNoSignalsTimeoutMillis, false); + this(defaultTimeoutMillis, defaultTimeoutMillis, defaultNoSignalsTimeoutMillis); } /** @@ -79,7 +116,7 @@ public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMi * @param defaultTimeoutMillis default timeout to be used in all expect* methods */ public TestEnvironment(long defaultTimeoutMillis) { - this(defaultTimeoutMillis, defaultTimeoutMillis, false); + this(defaultTimeoutMillis, defaultTimeoutMillis, defaultTimeoutMillis); } /** @@ -95,7 +132,7 @@ public TestEnvironment(long defaultTimeoutMillis) { * often helpful to pinpoint simple race conditions etc. */ public TestEnvironment(boolean printlnDebug) { - this(envDefaultTimeoutMillis(), envDefaultNoSignalsTimeoutMillis(), printlnDebug); + this(envDefaultTimeoutMillis(), envDefaultNoSignalsTimeoutMillis(), envDefaultPollTimeoutMillis(), printlnDebug); } /** @@ -124,6 +161,14 @@ public long defaultNoSignalsTimeoutMillis() { return defaultNoSignalsTimeoutMillis; } + /** + * The default amount of time to poll for events if {@code defaultTimeoutMillis} isn't preempted by an asynchronous + * event. + */ + public long defaultPollTimeoutMillis() { + return defaultPollTimeoutMillis; + } + /** * Tries to parse the env variable {@code DEFAULT_TIMEOUT_MILLIS} as long and returns the value if present OR its default value. * @@ -154,6 +199,21 @@ public static long envDefaultNoSignalsTimeoutMillis() { } } + /** + * Tries to parse the env variable {@code DEFAULT_POLL_TIMEOUT_MILLIS_ENV} as long and returns the value if present OR its default value. + * + * @throws java.lang.IllegalArgumentException when unable to parse the env variable + */ + public static long envDefaultPollTimeoutMillis() { + final String envMillis = System.getenv(DEFAULT_POLL_TIMEOUT_MILLIS_ENV); + if (envMillis == null) return envDefaultTimeoutMillis(); + else try { + return Long.parseLong(envMillis); + } catch (NumberFormatException ex) { + throw new IllegalArgumentException(String.format("Unable to parse %s env value [%s] as long!", DEFAULT_POLL_TIMEOUT_MILLIS_ENV, envMillis), ex); + } + } + /** * To flop means to "fail asynchronously", either by onErroring or by failing some TCK check triggered asynchronously. * This method does *NOT* fail the test - it's up to inspections of the error to fail the test if required. @@ -191,7 +251,7 @@ public void flop(Throwable thr, String msg) { asyncErrors.add(thr); } } - + /** * To flop means to "fail asynchronously", either by onErroring or by failing some TCK check triggered asynchronously. * This method does *NOT* fail the test - it's up to inspections of the error to fail the test if required. @@ -275,7 +335,7 @@ public Throwable dropAsyncError() { } /** - * Waits for {@link TestEnvironment#defaultTimeoutMillis()} and then verifies that no asynchronous errors + * Waits for {@link TestEnvironment#defaultNoSignalsTimeoutMillis()} and then verifies that no asynchronous errors * were signalled pior to, or during that time (by calling {@code flop()}). */ public void verifyNoAsyncErrors() { @@ -318,8 +378,13 @@ public void verifyNoAsyncErrorsNoDelay() { /** If {@code TestEnvironment#printlnDebug} is true, print debug message to std out. */ public void debug(String msg) { - if (printlnDebug) + if (debugEnabled()) { System.out.printf("[TCK-DEBUG] %s%n", msg); + } + } + + public final boolean debugEnabled() { + return printlnDebug; } /** @@ -512,26 +577,32 @@ public void expectCompletion(long timeoutMillis, String errorMsg) throws Interru } public void expectErrorWithMessage(Class expected, String requiredMessagePart) throws Exception { - expectErrorWithMessage(expected, requiredMessagePart, env.defaultTimeoutMillis()); + expectErrorWithMessage(expected, Collections.singletonList(requiredMessagePart), env.defaultTimeoutMillis(), env.defaultPollTimeoutMillis()); } public void expectErrorWithMessage(Class expected, List requiredMessagePartAlternatives) throws Exception { - expectErrorWithMessage(expected, requiredMessagePartAlternatives, env.defaultTimeoutMillis()); + expectErrorWithMessage(expected, requiredMessagePartAlternatives, env.defaultTimeoutMillis(), env.defaultPollTimeoutMillis()); } @SuppressWarnings("ThrowableResultOfMethodCallIgnored") public void expectErrorWithMessage(Class expected, String requiredMessagePart, long timeoutMillis) throws Exception { expectErrorWithMessage(expected, Collections.singletonList(requiredMessagePart), timeoutMillis); } + public void expectErrorWithMessage(Class expected, List requiredMessagePartAlternatives, long timeoutMillis) throws Exception { - final E err = expectError(expected, timeoutMillis); + expectErrorWithMessage(expected, requiredMessagePartAlternatives, timeoutMillis, timeoutMillis); + } + + public void expectErrorWithMessage(Class expected, List requiredMessagePartAlternatives, + long totalTimeoutMillis, long pollTimeoutMillis) throws Exception { + final E err = expectError(expected, totalTimeoutMillis, pollTimeoutMillis); final String message = err.getMessage(); - + boolean contains = false; - for (String requiredMessagePart : requiredMessagePartAlternatives) + for (String requiredMessagePart : requiredMessagePartAlternatives) if (message.contains(requiredMessagePart)) contains = true; // not short-circuting loop, it is expected to assertTrue(contains, - String.format("Got expected exception [%s] but missing message part [%s], was: %s", - err.getClass(), "anyOf: " + requiredMessagePartAlternatives, err.getMessage())); + String.format("Got expected exception [%s] but missing message part [%s], was: %s", + err.getClass(), "anyOf: " + requiredMessagePartAlternatives, err.getMessage())); } public E expectError(Class expected) throws Exception { @@ -539,7 +610,7 @@ public E expectError(Class expected) throws Exception { } public E expectError(Class expected, long timeoutMillis) throws Exception { - return expectError(expected, timeoutMillis, String.format("Expected onError(%s)", expected.getName())); + return expectError(expected, timeoutMillis, env.defaultPollTimeoutMillis()); } public E expectError(Class expected, String errorMsg) throws Exception { @@ -547,7 +618,16 @@ public E expectError(Class expected, String errorMsg) t } public E expectError(Class expected, long timeoutMillis, String errorMsg) throws Exception { - return received.expectError(expected, timeoutMillis, errorMsg); + return expectError(expected, timeoutMillis, env.defaultPollTimeoutMillis(), errorMsg); + } + + public E expectError(Class expected, long totalTimeoutMillis, long pollTimeoutMillis) throws Exception { + return expectError(expected, totalTimeoutMillis, pollTimeoutMillis, String.format("Expected onError(%s)", expected.getName())); + } + + public E expectError(Class expected, long totalTimeoutMillis, long pollTimeoutMillis, + String errorMsg) throws Exception { + return received.expectError(expected, totalTimeoutMillis, pollTimeoutMillis, errorMsg); } public void expectNone() throws InterruptedException { @@ -576,7 +656,9 @@ public ManualSubscriberWithSubscriptionSupport(TestEnvironment env) { @Override public void onNext(T element) { - env.debug(String.format("%s::onNext(%s)", this, element)); + if (env.debugEnabled()) { + env.debug(String.format("%s::onNext(%s)", this, element)); + } if (subscription.isCompleted()) { super.onNext(element); } else { @@ -586,7 +668,9 @@ public void onNext(T element) { @Override public void onComplete() { - env.debug(this + "::onComplete()"); + if (env.debugEnabled()) { + env.debug(this + "::onComplete()"); + } if (subscription.isCompleted()) { super.onComplete(); } else { @@ -596,7 +680,9 @@ public void onComplete() { @Override public void onSubscribe(Subscription s) { - env.debug(String.format("%s::onSubscribe(%s)", this, s)); + if (env.debugEnabled()) { + env.debug(String.format("%s::onSubscribe(%s)", this, s)); + } if (!subscription.isCompleted()) { subscription.complete(s); } else { @@ -606,7 +692,9 @@ public void onSubscribe(Subscription s) { @Override public void onError(Throwable cause) { - env.debug(String.format("%s::onError(%s)", this, cause)); + if (env.debugEnabled()) { + env.debug(String.format("%s::onError(%s)", this, cause)); + } if (subscription.isCompleted()) { super.onError(cause); } else { @@ -629,7 +717,9 @@ public BlackholeSubscriberWithSubscriptionSupport(TestEnvironment env) { @Override public void onNext(T element) { - env.debug(String.format("%s::onNext(%s)", this, element)); + if (env.debugEnabled()) { + env.debug(String.format("%s::onNext(%s)", this, element)); + } if (!subscription.isCompleted()) { env.flop(String.format("Subscriber::onNext(%s) called before Subscriber::onSubscribe", element)); } @@ -801,7 +891,7 @@ public void expectCancelling() throws InterruptedException { public void expectCancelling(long timeoutMillis) throws InterruptedException { cancelled.expectClose(timeoutMillis, "Did not receive expected cancelling of upstream subscription"); } - + public boolean isCancelled() throws InterruptedException { return cancelled.isClosed(); } @@ -882,11 +972,12 @@ public Promise(TestEnvironment env) { } private ArrayBlockingQueue abq = new ArrayBlockingQueue(1); - private volatile T _value = null; + private AtomicReference _value = new AtomicReference(); public T value() { - if (isCompleted()) { - return _value; + final T value = _value.get(); + if (value != null) { + return value; } else { env.flop("Cannot access promise value before completion"); return null; @@ -894,22 +985,28 @@ public T value() { } public boolean isCompleted() { - return _value != null; + return _value.get() != null; } /** * Allows using expectCompletion to await for completion of the value and complete it _then_ */ public void complete(T value) { - abq.add(value); + if (_value.compareAndSet(null, value)) { + // we add the value to the queue such to wake up any expectCompletion which was triggered before complete() was called + abq.add(value); + } else { + env.flop(String.format("Cannot complete a promise more than once! Present value: %s, attempted to set: %s", _value.get(), value)); + } } /** - * Completes the promise right away, it is not possible to expectCompletion on a Promise completed this way + * Same as complete. + * + * Keeping this method for binary compatibility. */ public void completeImmediatly(T value) { - complete(value); // complete! - _value = value; // immediatly! + complete(value); } public void expectCompletion(long timeoutMillis, String errorMsg) throws InterruptedException { @@ -918,8 +1015,6 @@ public void expectCompletion(long timeoutMillis, String errorMsg) throws Interru if (val == null) { env.flop(String.format("%s within %d ms", errorMsg, timeoutMillis)); - } else { - _value = val; } } } @@ -1003,22 +1098,44 @@ public void expectCompletion(long timeoutMillis, String errorMsg) throws Interru } // else, ok } - @SuppressWarnings("unchecked") + /** + * @deprecated Deprecated in favor of {@link #expectError(Class, long, long, String)}. + */ + @Deprecated public E expectError(Class clazz, long timeoutMillis, String errorMsg) throws Exception { - Thread.sleep(timeoutMillis); - - if (env.asyncErrors.isEmpty()) { - return env.flopAndFail(String.format("%s within %d ms", errorMsg, timeoutMillis)); - } else { - // ok, there was an expected error - Throwable thrown = env.asyncErrors.remove(0); + return expectError(clazz, timeoutMillis, timeoutMillis, errorMsg); + } - if (clazz.isInstance(thrown)) { - return (E) thrown; + @SuppressWarnings("unchecked") + final E expectError(Class clazz, final long totalTimeoutMillis, + long pollTimeoutMillis, + String errorMsg) throws Exception { + long totalTimeoutRemainingNs = MILLISECONDS.toNanos(totalTimeoutMillis); + long timeStampANs = System.nanoTime(); + long timeStampBNs; + + for (;;) { + Thread.sleep(Math.min(pollTimeoutMillis, NANOSECONDS.toMillis(totalTimeoutRemainingNs))); + + if (env.asyncErrors.isEmpty()) { + timeStampBNs = System.nanoTime(); + totalTimeoutRemainingNs =- timeStampBNs - timeStampANs; + timeStampANs = timeStampBNs; + + if (totalTimeoutRemainingNs <= 0) { + return env.flopAndFail(String.format("%s within %d ms", errorMsg, totalTimeoutMillis)); + } } else { + // ok, there was an expected error + Throwable thrown = env.asyncErrors.remove(0); - return env.flopAndFail(String.format("%s within %d ms; Got %s but expected %s", - errorMsg, timeoutMillis, thrown.getClass().getCanonicalName(), clazz.getCanonicalName())); + if (clazz.isInstance(thrown)) { + return (E) thrown; + } else { + + return env.flopAndFail(String.format("%s within %d ms; Got %s but expected %s", + errorMsg, totalTimeoutMillis, thrown.getClass().getCanonicalName(), clazz.getCanonicalName())); + } } } } diff --git a/tck/src/test/java/org/reactivestreams/tck/PublisherVerificationTest.java b/tck/src/test/java/org/reactivestreams/tck/PublisherVerificationTest.java index ee4dc9b6..ca42f00c 100644 --- a/tck/src/test/java/org/reactivestreams/tck/PublisherVerificationTest.java +++ b/tck/src/test/java/org/reactivestreams/tck/PublisherVerificationTest.java @@ -228,6 +228,39 @@ public void optional_spec105_emptyStreamMustTerminateBySignallingOnComplete_shou }, "Subscriber::onComplete() called before Subscriber::onSubscribe"); } + @Test + public void optional_spec105_emptyStreamMustTerminateBySignallingOnComplete_shouldPass_whenOnCompleteSynchronouslyInovked() throws Throwable { + final Publisher synchronousOnCompletePublisher = new Publisher() { + @Override public void subscribe(final Subscriber s) { + s.onSubscribe(new Subscription() { + @Override public void request(long n) { } + @Override public void cancel() { } + }); + s.onComplete(); + } + }; + + requireOptionalTestPass(new ThrowingRunnable() { + @Override public void run() throws Throwable { + PublisherVerification verification = new PublisherVerification(newTestEnvironment()) { + @Override public Publisher createPublisher(long elements) { + return synchronousOnCompletePublisher; + } + + @Override public long maxElementsFromPublisher() { + return 0; // it is an "empty" Publisher + } + + @Override public Publisher createFailedPublisher() { + return null; + } + }; + + verification.optional_spec105_emptyStreamMustTerminateBySignallingOnComplete(); + } + }); + } + @Test public void required_spec107_mustNotEmitFurtherSignalsOnceOnCompleteHasBeenSignalled_shouldFailForNotCompletingPublisher() throws Throwable { requireTestFailure(new ThrowingRunnable() { diff --git a/tck/src/test/java/org/reactivestreams/tck/SubscriberWhiteboxVerificationTest.java b/tck/src/test/java/org/reactivestreams/tck/SubscriberWhiteboxVerificationTest.java index bd0737d7..0441e6ed 100644 --- a/tck/src/test/java/org/reactivestreams/tck/SubscriberWhiteboxVerificationTest.java +++ b/tck/src/test/java/org/reactivestreams/tck/SubscriberWhiteboxVerificationTest.java @@ -23,7 +23,10 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; /** * Validates that the TCK's {@link SubscriberWhiteboxVerification} fails with nice human readable errors. @@ -31,8 +34,8 @@ */ public class SubscriberWhiteboxVerificationTest extends TCKVerificationSupport { - private ExecutorService ex; - @BeforeClass void before() { ex = Executors.newFixedThreadPool(4); } + private ScheduledExecutorService ex; + @BeforeClass void before() { ex = Executors.newScheduledThreadPool(4); } @AfterClass void after() { if (ex != null) ex.shutdown(); } @Test @@ -217,6 +220,51 @@ public Subscriber apply(WhiteboxSubscriberProbe probe) throws }, "But I thought it's cancelled!"); } + @Test + public void required_spec208_mustBePreparedToReceiveOnNextSignalsAfterHavingCalledSubscriptionCancel_shouldWaitForDemandBeforeSignalling() throws Throwable { + customSubscriberVerification(new Function, Subscriber>() { + @Override + public Subscriber apply(WhiteboxSubscriberProbe probe) throws Throwable { + + final AtomicBoolean demandRequested = new AtomicBoolean(false); + + return new SimpleSubscriberWithProbe(probe) { + @Override public void onSubscribe(final Subscription s) { + this.subscription = s; + probe.registerOnSubscribe(new SubscriberPuppet() { + @Override public void triggerRequest(final long elements) { + ex.schedule(new Runnable() { + @Override + public void run() { + demandRequested.set(true); + subscription.request(elements); + } + }, TestEnvironment.envDefaultTimeoutMillis() / 2, TimeUnit.MILLISECONDS); + } + + @Override public void signalCancel() { + // Delay this too to ensure that cancel isn't invoked before request. + ex.schedule(new Runnable() { + @Override + public void run() { + subscription.cancel(); + } + }, TestEnvironment.envDefaultTimeoutMillis() / 2, TimeUnit.MILLISECONDS); + } + }); + } + + @Override public void onNext(Integer element) { + if (!demandRequested.get()) { + throw new RuntimeException("onNext signalled without demand!"); + } + probe.registerOnNext(element); + } + }; + } + }).required_spec208_mustBePreparedToReceiveOnNextSignalsAfterHavingCalledSubscriptionCancel(); + } + @Test public void required_spec209_mustBePreparedToReceiveAnOnCompleteSignalWithPrecedingRequestCall_shouldFail() throws Throwable { requireTestFailure(new ThrowingRunnable() { @@ -299,7 +347,7 @@ public Subscriber apply(WhiteboxSubscriberProbe probe) throws } @Test - public void required_spec308_requestMustRegisterGivenNumberElementsToBeProduced_shouldFail() throws Throwable { + public void required_spec308_requestMustRegisterGivenNumberElementsToBeProduced_shouldPass() throws Throwable { // sanity checks the "happy path", that triggerRequest() propagates the right demand customSubscriberVerification(new Function, Subscriber>() { @Override @@ -309,6 +357,114 @@ public Subscriber apply(WhiteboxSubscriberProbe probe) throws }).required_spec308_requestMustRegisterGivenNumberElementsToBeProduced(); } + @Test + public void required_spec308_requestMustRegisterGivenNumberElementsToBeProduced_shouldWaitForDemandBeforeSignalling() throws Throwable { + customSubscriberVerification(new Function, Subscriber>() { + @Override + public Subscriber apply(WhiteboxSubscriberProbe probe) throws Throwable { + + final AtomicBoolean demandRequested = new AtomicBoolean(false); + return new SimpleSubscriberWithProbe(probe) { + @Override + public void onSubscribe(Subscription s) { + this.subscription = s; + probe.registerOnSubscribe(new SubscriberPuppet() { + @Override + public void triggerRequest(final long elements) { + ex.schedule(new Runnable() { + @Override + public void run() { + demandRequested.set(true); + subscription.request(elements); + } + }, TestEnvironment.envDefaultTimeoutMillis() / 2, TimeUnit.MILLISECONDS); + } + + @Override + public void signalCancel() { + // Delay this too to ensure that cancel isn't invoked before request. + ex.schedule(new Runnable() { + @Override + public void run() { + subscription.cancel(); + } + }, TestEnvironment.envDefaultTimeoutMillis() / 2, TimeUnit.MILLISECONDS); + } + }); + } + + @Override + public void onNext(Integer element) { + if (!demandRequested.get()) { + throw new RuntimeException("onNext signalled without demand!"); + } + probe.registerOnNext(element); + } + }; + } + }).required_spec308_requestMustRegisterGivenNumberElementsToBeProduced(); + } + + @Test + public void required_spec308_requestMustRegisterGivenNumberElementsToBeProduced_shouldWaitForDemandTwiceForOneAtATimeSubscribers() throws Throwable { + customSubscriberVerification(new Function, Subscriber>() { + @Override + public Subscriber apply(WhiteboxSubscriberProbe probe) throws Throwable { + + final AtomicLong outstandingRequest = new AtomicLong(0); + final AtomicBoolean demandRequested = new AtomicBoolean(); + return new SimpleSubscriberWithProbe(probe) { + @Override + public void onSubscribe(Subscription s) { + this.subscription = s; + probe.registerOnSubscribe(new SubscriberPuppet() { + @Override + public void triggerRequest(final long elements) { + outstandingRequest.getAndAdd(elements); + ex.schedule(new Runnable() { + @Override + public void run() { + demandRequested.set(true); + subscription.request(1); + } + }, TestEnvironment.envDefaultTimeoutMillis() / 2, TimeUnit.MILLISECONDS); + } + + @Override + public void signalCancel() { + // Delay this too to ensure that cancel isn't invoked before request. + ex.schedule(new Runnable() { + @Override + public void run() { + subscription.cancel(); + } + }, TestEnvironment.envDefaultTimeoutMillis() / 2, TimeUnit.MILLISECONDS); + } + }); + } + + @Override + public void onNext(Integer element) { + if (!demandRequested.getAndSet(false)) { + throw new RuntimeException("onNext signalled without demand!"); + } + if (outstandingRequest.decrementAndGet() > 0) { + ex.schedule(new Runnable() { + @Override + public void run() { + demandRequested.set(true); + subscription.request(1); + } + }, TestEnvironment.envDefaultTimeoutMillis() / 2, TimeUnit.MILLISECONDS); + } + probe.registerOnNext(element); + } + }; + } + }).required_spec308_requestMustRegisterGivenNumberElementsToBeProduced(); + } + + // FAILING IMPLEMENTATIONS // /** diff --git a/tck/src/test/java/org/reactivestreams/tck/flow/support/TCKVerificationSupport.java b/tck/src/test/java/org/reactivestreams/tck/flow/support/TCKVerificationSupport.java index 5251d231..a7fd730c 100644 --- a/tck/src/test/java/org/reactivestreams/tck/flow/support/TCKVerificationSupport.java +++ b/tck/src/test/java/org/reactivestreams/tck/flow/support/TCKVerificationSupport.java @@ -81,6 +81,18 @@ public void requireTestSkip(ThrowingRunnable run, String msgPart) { throw new RuntimeException("Expected TCK to SKIP this test, instead if PASSed!"); } + public void requireOptionalTestPass(ThrowingRunnable run) { + try { + run.run(); + } catch (SkipException skip) { + throw new RuntimeException("Expected TCK to PASS this test, instead it was SKIPPED", skip.getCause()); + } catch (Throwable throwable) { + throw new RuntimeException( + String.format("Expected TCK to PASS this test, yet it threw %s(%s) instead!", + throwable.getClass().getName(), throwable.getMessage()), throwable); + } + } + /** * This publisher does NOT fulfil all Publisher spec requirements. * It's just the bare minimum to enable this test to fail the Subscriber tests.