Merge pull request reactive-streams#14 from typesafehub/wip-apispi-in-java-√

Reimplements the SPI and API in Java

Author: viktorklang

spi/build.sbt

@@ -1,3 +1,13 @@
 name := "reactive-streams-spi"
 
+javacOptions in compile ++= Seq("-encoding", "UTF-8", "-source", "1.6", "-target", "1.6", "-Xlint:unchecked", "-Xlint:deprecation")
+
+javacOptions in (Compile,doc) ++= Seq("-encoding","UTF-8","-docencoding", "UTF-8", "-charset", "UTF-8", "-notimestamp", "-linksource")
+
+autoScalaLibrary := false
+
+crossPaths := false
+
+publishMavenStyle := true
+
 Common.javadocSettings

spi/src/main/java/org/reactivestreams/api/Consumer.java

@@ -0,0 +1,21 @@
+package org.reactivestreams.api;
+
+import org.reactivestreams.spi.Subscriber;
+
+/**
+ * A Consumer is the logical sink of elements of a given type.
+ * The underlying implementation is done by way of a {@link org.reactivestreams.spi.Subscriber Subscriber}.
+ * This interface is the user-level API for a sink while a Subscriber is the SPI.
+ * <p>
+ * Implementations of this interface will typically offer domain- or language-specific
+ * methods for transforming or otherwise interacting with the stream of elements.
+ */
+public interface Consumer<T> {
+  
+  /**
+   * Get the underlying {@link org.reactivestreams.spi.Subscriber Subscriber} for this Consumer. This method should only be used by
+   * implementations of this API.
+   * @return the underlying subscriber for this consumer
+   */
+  public Subscriber<T> getSubscriber();
+}

spi/src/main/scala/org/reactivestreams/api/Processor.scala renamed to spi/src/main/java/org/reactivestreams/api/Processor.java

@@ -1,9 +1,10 @@
-package org.reactivestreams.api
+package org.reactivestreams.api;
 
 /**
  * A Processor is a stand-alone representation of a transformation for
  * elements from In to Out types. Implementations of this API will provide
  * factory methods for creating Processors and connecting them to
- * [[Producer]] and [[Consumer]].
+ * {@link org.reactivestreams.api.Producer Producer} and {@link org.reactivestreams.api.Consumer Consumer}.
  */
-trait Processor[In, Out] extends Consumer[In] with Producer[Out]
+public interface Processor<I, O> extends Consumer<I>, Producer<O> {
+}

spi/src/main/java/org/reactivestreams/api/Producer.java

@@ -0,0 +1,38 @@
+package org.reactivestreams.api;
+
+import org.reactivestreams.spi.Publisher;
+
+/**
+ * A Producer is the logical source of elements of a given type. 
+ * The underlying implementation is done by way of a {@link org.reactivestreams.spi.Publisher Publisher}.
+ * This interface is the user-level API for a source while a Publisher is the
+ * SPI.
+ * <p>
+ * Implementations of this interface will typically offer domain- or language-specific
+ * methods for transforming or otherwise interacting with the produced stream of elements.
+ */
+public interface Producer<T> {
+  
+  /**
+   * Get the underlying {@link org.reactivestreams.spi.Publisher Publisher} for this Producer. This method should only be used by
+   * implementations of this API.
+   * @return the underlying publisher for this producer
+   */
+  public Publisher<T> getPublisher();
+  
+  /**
+   * Connect the given consumer to this producer. This means that the
+   * Subscriber underlying the {@link org.reactivestreams.api.Consumer Consumer} subscribes to this Producer’s
+   * underlying {@link org.reactivestreams.spi.Publisher Publisher}, which will initiate the transfer of the produced
+   * stream of elements from producer to consumer until either of three things
+   * happen:
+   * <p>
+   * <ul>
+   * <li>The stream ends normally (no more elements available).</li>
+   * <li>The producer encounters a fatal error condition.</li>
+   * <li>The consumer cancels the reception of more elements.</li>
+   * </ul>
+   * @param consumer The consumer to register with this producer.
+   */
+  public void produceTo(Consumer<T> consumer);
+}

spi/src/main/java/org/reactivestreams/spi/Publisher.java

@@ -0,0 +1,16 @@
+package org.reactivestreams.spi;
+
+/**
+ * A Publisher is a source of elements of a given type. One or more {@link org.reactivestreams.spi.Subscriber Subscriber} may be connected
+ * to this Publisher in order to receive the published elements, contingent on availability of these
+ * elements as well as the presence of demand signaled by the Subscriber via {@link org.reactivestreams.spi.Subscription#requestMore(int) requestMore}.
+ */
+public interface Publisher<T> {
+  
+  /**
+   * Subscribe the given {@link org.reactivestreams.spi.Subscriber Subscriber} to this Publisher. A Subscriber can at most be subscribed once
+   * to a given Publisher, and to at most one Publisher in total.
+   * @param subscriber The subscriber to register with this publisher.
+   */
+  public void subscribe(Subscriber<T> subscriber);
+}

spi/src/main/java/org/reactivestreams/spi/Subscriber.java

@@ -0,0 +1,43 @@
+package org.reactivestreams.spi;
+
+/**
+ * A Subscriber receives elements from a {@link org.reactivestreams.spi.Publisher Publisher} based on the {@link org.reactivestreams.spi.Subscription Subscription} it has.
+ * The Publisher may supply elements as they become available, the Subscriber signals demand via 
+ * {@link org.reactivestreams.spi.Subscription#requestMore(int) requestMore} and elements from when supply and demand are both present.
+ */
+public interface Subscriber<T> {
+  
+  /**
+   * The {@link org.reactivestreams.spi.Publisher Publisher} generates a {@link org.reactivestreams.spi.Subscription Subscription} upon {@link org.reactivestreams.spi.Publisher#subscribe(org.reactivestreams.spi.Subscriber) subscribe} and passes
+   * it on to the Subscriber named there using this method. The Publisher may choose to reject
+   * the subscription request by calling {@link #onError onError} instead.
+   * @param subscription The subscription which connects this subscriber to its publisher.
+   */
+  public void onSubscribe(Subscription subscription);
+  
+  /**
+   * The {@link org.reactivestreams.spi.Publisher Publisher} calls this method to pass one element to this Subscriber. The element
+   * must not be <code>null</code>. The Publisher must not call this method more often than
+   * the Subscriber has signaled demand for via the corresponding {@link org.reactivestreams.spi.Subscription Subscription}.
+   * @param element The element that is passed from publisher to subscriber.
+   */
+  public void onNext(T element);
+  
+  /**
+   * The {@link org.reactivestreams.spi.Publisher Publisher} calls this method in order to signal that it terminated normally.
+   * No more elements will be forthcoming and none of the Subscriber’s methods will be called hereafter.
+   */
+  public void onComplete();
+  
+  /**
+   * The {@link org.reactivestreams.spi.Publisher Publisher} calls this method to signal that the stream of elements has failed
+   * and is being aborted. The Subscriber should abort its processing as soon as possible.
+   * No more elements will be forthcoming and none of the Subscriber’s methods will be called hereafter.
+   * <p>
+   * This method is not intended to pass validation errors or similar from Publisher to Subscriber
+   * in order to initiate an orderly shutdown of the exchange; it is intended only for fatal
+   * failure conditions which make it impossible to continue processing further elements.
+   * @param cause An exception which describes the reason for tearing down this stream.
+   */
+  public void onError(Throwable cause);
+}

spi/src/main/java/org/reactivestreams/spi/Subscription.java

@@ -0,0 +1,26 @@
+package org.reactivestreams.spi;
+
+/**
+ * A Subscription models the relationship between a {@link org.reactivestreams.spi.Publisher Publisher} and a {@link org.reactivestreams.spi.Subscriber Subscriber}.
+ * The Subscriber receives a Subscription so that it can ask for elements to be delivered
+ * using {@link org.reactivestreams.spi.Subscription#requestMore(int) requestMore}. The Subscription can be disposed of by canceling it.
+ */
+public interface Subscription {
+  
+  /**
+   * Cancel this subscription. The {@link org.reactivestreams.spi.Publisher Publisher} to which produced this Subscription 
+   * will eventually stop sending more elements to the {@link org.reactivestreams.spi.Subscriber Subscriber} which owns
+   * this Subscription. This may happen before the requested number of elements has
+   * been delivered, even if the Publisher would still have more elements.
+   */
+  public void cancel();
+  
+  /**
+   * Request more data from the {@link org.reactivestreams.spi.Publisher Publisher} which produced this Subscription.
+   * The number of requested elements is cumulative to the number requested previously.
+   * The Publisher may eventually publish up to the requested number of elements to
+   * the {@link org.reactivestreams.spi.Subscriber Subscriber} which owns this Subscription.
+   * @param elements The number of elements requested.
+   */
+  public void requestMore(int elements);
+}