review() : Fix some typo and refactor for publication

- Change method name
- Update README
- Change demo Main to use all callbacks

Author: bordigoni

README.adoc

@@ -2,22 +2,37 @@
 == Streamdata.io Java SDK
 
 
-=== What does the API do ?
+=== What does the API do?
 
 It remove the plumbing between you and streamdata.io once you have an API ready to use, you can use the code above.
 
 It connects to the server with some your API URL and your credentials (API key) and you get notified using either
 
 * Some callback you provide
-* Trough an RxJava2 `Flowable`
+* Through an RxJava2 `Flowable`
 
 You can choose to receive snapshots only or one snapshot and then patches you don't even know SSE is used, you just use your data.
 
-=== Prerequisite
+=== Step by step setup to run demo
+
+* Create an free account on streamdata.io https://portal.streamdata.io/#/register to get an App token.
+* Clone project, edit Main.java and replace [YOUR TOKEN HERE] with your App token.
+* Make sure you have Java 8+ installed
+* Make sure you have maven 3.0+ installed
+* Build project with maven:
+
+    `mvn clean install`
+
+* Run sample from a terminal:
+
+    `java -jar target/streamdataio-sdk-java-1.0.jar`
+
+* You should see data and patches pushed in your application and displayed on your terminal.
+* You can build the jar on you own and use provided classes to use this API
+you can use the provided demo example API which simulates updating stocks prices from a financial market: 'http://stockmarket.streamdata.io/v2/prices'
+
+Feel free to test it with any REST/Json API of your choice.
 
-* You have to create an account on streamdata.io and create an app using your API or one of demo sample API provided by streamdata.io
-* You must use Java8+ to use this API as to takes advantage
-of lambda expression and some specific features of java8.
 
 === How to integrate the API into your project
 
@@ -42,9 +57,11 @@ dependencies {
 
 Notable transitive dependencies :
 
-* Glassfish Jersey as EventSource client
+* Glassfish Jersey as EventSource client (https://jersey.java.net/documentation/latest/sse.html)
+* RxJava 2 (https://github.com/ReactiveX/RxJava)
+* Jackson for Json processing  (http://wiki.fasterxml.com/JacksonHome).
+* zjsonpatch (https://github.com/flipkart-incubator/zjsonpatch) as a Java Json-Patch implementation
 * Slf4j with a default configuration
-* RxJava 2
 
 === Callback based API
 
@@ -58,7 +75,7 @@ String appKey = "YOUR OWN APP KEY";
 
 EventSourceClient client = StreamdataClient.createClient(apiURL, appKey);
 client.addHeader("X-MYAPI-POLLER", "Polled By SD.io")                           # <2>
-        .incrementalCache(true)                                                 # <3>
+        .useJsonPatch(true)                                                     # <3>
         .onSnapshot(data -> this::processData)                                  # <4>
         .onPatch(patch -> this.processPatch(patch, client.getCurrentSnaphot())) # <5>
         .onOpen(() -> logger.info("And we are... live!"))                       # <6>
@@ -71,12 +88,12 @@ client.addHeader("X-MYAPI-POLLER", "Polled By SD.io")
 <1> This the key you when you created your app in the web portal.
 <2> Headers to be added when polling your API (Typically some auth header) [_Optional_ ]
 <3> This allows you let streamdata send snapshots only (set `false` to do so) or snapshot then patch (default behaviour if you don't call this method) [_Optional_ ]
-<4> A callback that will consume the snaphost
+<4> A callback that will consume the snapshot
 <5> A callback that will consume the patch (you can access the snaphost anyways) [_Optional_ ]
 <6> Call when successfully open the connection [_Optional_ ]
 <7> Call the connection is actually closed [_Optional_ ]
-<6> Call when the stream return an error, by default the error is logged usually it is recommended to close the connection : ```client.close()``` [_Optional_ ]
-<7> Call when an error occurs, mainly when opening the event source, by default the error is recommended to close the connection : ```client.close()``` [_Optional_ ]
+<6> Call when the stream return an error, by default the error is logged usually it is recommended to close the connection: ```client.close()``` [_Optional_ ]
+<7> Call when an error occurs, mainly when opening the event source, by default the error is recommended to close the connection: ```client.close()``` [_Optional_ ]
 <10> Open the connection with the server
 
 
@@ -95,7 +112,7 @@ String appKey = "YOUR OWN APP KEY";
 
 StreamdataClient.createRxJavaClient(apiURL, appKey)
             .addHeader("X-MYAPI-POLLER", "Polled By SD.io")                     # <2>
-            .incrementalCache(true)                                             # <3>
+            .useJsonPatch(true)                                             # <3>
             .toFlowable()                                                       # <4>
             // here you can add operator to manipulate the flow
             .subscribe(event -> {                                               # <5>
@@ -118,9 +135,9 @@ StreamdataClient.createRxJavaClient(apiURL, appKey)
 
 == Errors
 
-Errors not a simple string. It is JSON !
+Errors not a simple string. It is JSON!
 
-Above, an example of an error so you can get more detailed informations
+Above, an example of an error so you can get more detailed informations.
 
 [JSON]
 ```

pom.xml

@@ -4,7 +4,7 @@
 
     <groupId>io.streamdata</groupId>
     <artifactId>streamdataio-java-sdk</artifactId>
-    <version>0.0.1-SNAPSHOT</version>
+    <version>1.0</version>
     <packaging>jar</packaging>
 
     <name>Streamdata.io Java SDK</name>
@@ -73,6 +73,64 @@
                     <showWarnings>true</showWarnings>
                 </configuration>
             </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <version>2.6</version>
+                <configuration>
+                    <archive>
+                        <manifest>
+                            <mainClass>io.streamdata.demo.Main</mainClass>
+                        </manifest>
+                    </archive>
+                    <descriptorRefs>
+                        <descriptorRef>jar-with-dependencies</descriptorRef>
+                    </descriptorRefs>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>make-assembly</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-dependency-plugin</artifactId>
+                <version>2.10</version>
+                <executions>
+                    <execution>
+                        <id>copy-dependencies</id>
+                        <phase>prepare-package</phase>
+                        <goals>
+                            <goal>copy-dependencies</goal>
+                        </goals>
+                        <configuration>
+                            <outputDirectory>${project.build.directory}/lib</outputDirectory>
+                            <overWriteReleases>false</overWriteReleases>
+                            <overWriteSnapshots>false</overWriteSnapshots>
+                            <overWriteIfNewer>true</overWriteIfNewer>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <version>2.6</version>
+                <configuration>
+                    <archive>
+                        <manifest>
+                            <addClasspath>true</addClasspath>
+                            <classpathPrefix>lib/</classpathPrefix>
+                            <mainClass>io.streamdata.demo.Main</mainClass>
+                        </manifest>
+                    </archive>
+                </configuration>
+            </plugin>
         </plugins>
     </build>
 

src/main/java/io/streamdata/demo/Main.java

@@ -13,7 +13,7 @@ public class Main {
 
     public static void main(String... args) throws URISyntaxException, InterruptedException {
 
-        final String appKey = "ODZjZDQ5MDYtYzZkYS00NTQwLWI0ZDctMGZlYzU2N2JlYmY3";
+        final String appKey = "[YOUR TOKEN HERE]";
         final String apiURL = "http://stockmarket.streamdata.io/prices";
 
         Logger logger = LoggerFactory.getLogger(Main.class);
@@ -27,8 +27,10 @@ public static void main(String... args) throws URISyntaxException, InterruptedEx
             eventSource
                     .addHeader("X-MYAPI-HEADER", "Polled-By-SD.io")
                     .addHeader("X-MYAPI-HEADER2", "SomeStuffs")
-                    .onSnapshot(data -> logger.info("INITIAL SNAPSHOT {}", data))
+                    .onSnapshot(data -> logger.info("RECEIVING SNAPSHOT {}", data))
                     .onPatch(patch -> logger.info("PATCH {} SNAPSHOT UPDATED {}", patch, eventSource.getCurrentSnapshot()))
+                    .onError(error -> logger.error("Error occured : {}", error.toString()))
+                    .onException(ex -> logger.error("Error occured on connection", ex))
                     .onOpen(() -> logger.info("And we are... live!"))
                     .onClose(() -> logger.info("Bye now!"))
                     .open();
@@ -44,8 +46,8 @@ public static void main(String... args) throws URISyntaxException, InterruptedEx
         {
 
             EventSourceClient eventSource = StreamdataClient.createClient(apiURL, appKey)
-                    .incrementalCache(false)
-                    .onSnapshot(data -> logger.info("INITIAL SNAPSHOT {}", data))
+                    .useJsonPatch(false)
+                    .onSnapshot(data -> logger.info("RECEIVING SNAPSHOT {}", data))
                     .onOpen(() -> logger.info("And we are... live!"))
                     .open();
 
@@ -61,7 +63,7 @@ public static void main(String... args) throws URISyntaxException, InterruptedEx
             RxJavaEventSourceClient rxJavaEventSourceClient = StreamdataClient.createRxJavaClient(apiURL, appKey);
             Disposable disposable =
                     rxJavaEventSourceClient.addHeader("X-MYAPI-HEADER", "Polled By SD.io")
-                            .incrementalCache(true) // same behavior as default
+                            .useJsonPatch(true) // same behavior as default
                             .toFlowable()
                             .subscribe(event -> {
                                 if (event.isSnapshot()) {

src/main/java/io/streamdata/sdk/EventSourceClient.java

@@ -27,9 +27,9 @@ public interface EventSourceClient {
      * <p>If set to false a snapshot will be sent every time, no patch is sent. This means that {@link RxJavaEventSourceClient.Event#getPatch()} will return null <b>Use this only for low frequency polling</b></p>
      * <p>Behind the scene it adds the header <code>text/event-stream</code> for patches or <code>application/json</code> for non-incremental cache</p>
      *
-     * @param enableIncrementalCache a boolean to allow incremental cache (default : true)
+     * @param useJsonPatch a boolean to allow incremental cache (default : true)
      */
-    EventSourceClient incrementalCache(boolean enableIncrementalCache);
+    EventSourceClient useJsonPatch(boolean useJsonPatch);
 
     /**
      * Sets a optionnal callback to be called after the event source has been successfully started.
@@ -96,7 +96,7 @@ public interface EventSourceClient {
 
     /**
      * Opens the connections with streamdata proxy that will poll data for you.
-     * {@link #onSnapshot(Consumer)} must have called before and {@link #onPatch(Consumer)} is incremental cache is on (default is yes unless you call {@link #incrementalCache(boolean)} with false.
+     * {@link #onSnapshot(Consumer)} must have called before and {@link #onPatch(Consumer)} is incremental cache is on (default is yes unless you call {@link #useJsonPatch(boolean)} with false.
      *
      * @return a future to get hints on the thread status
      */

src/main/java/io/streamdata/sdk/RxJavaEventSourceClient.java

@@ -40,9 +40,9 @@ public interface RxJavaEventSourceClient {
      * <p>If set to false a snapshot will be sent every time, no patch is sent. This means that {@link Event#getPatch()} will return null <b>Use this only for low frequency polling</b></p>
      * <p>Behind the scene it adds the header <code>text/event-stream</code> for patches or <code>application/json</code> for non-incremental cache</p>
      *
-     * @param enableIncrementalCache a boolean to allow incremental cache (default : true)
+     * @param useJsonPatch a boolean to allow incremental cache (default : true)
      */
-    RxJavaEventSourceClient incrementalCache(boolean enableIncrementalCache);
+    RxJavaEventSourceClient useJsonPatch(boolean useJsonPatch);
 
 
     /**
@@ -121,7 +121,7 @@ public JsonNode getSnapshot() {
          * Gets the patch if any. There are two cases where <b>the patch can be null</b>
          * <ul>
          * <li>{@link #isSnapshot()} return true</li>
-         * <li>{@link EventSourceClient#incrementalCache(boolean)} or {@link RxJavaEventSourceClient#incrementalCache(boolean)} has been called with null</li>
+         * <li>{@link EventSourceClient#useJsonPatch(boolean)} or {@link RxJavaEventSourceClient#useJsonPatch(boolean)} has been called with null</li>
          * </ul>
          *
          * @return the as a JsonNode

src/main/java/io/streamdata/sdk/impl/EventSourceClientImpl.java

@@ -40,7 +40,7 @@ public class EventSourceClientImpl implements EventSourceClient {
     private Consumer<JsonNode> onDataCallback;
     private Consumer<JsonNode> onPatchCallback;
     private Consumer<JsonNode> onErrorCallback = err -> LOGGER.error("A streamdata error has been sent from SSE : {}", err);
-    private Consumer<Throwable> onFailureCallback = t -> LOGGER.error("An error occured while processing event", t);
+    private Consumer<Throwable> onExceptionCallback = t -> LOGGER.error("An error occured while processing event", t);
 
     // jackson objectMapper to parse Json content
     private final ObjectMapper jsonObjectMapper = new ObjectMapper();
@@ -93,7 +93,7 @@ public EventSourceClient addHeader(String name, String value) {
     }
 
     @Override
-    public EventSourceClient incrementalCache(boolean enableIncrementalCache) {
+    public EventSourceClient useJsonPatch(boolean enableIncrementalCache) {
         this.incrementalCache = enableIncrementalCache;
         return this;
     }
@@ -130,7 +130,7 @@ public EventSourceClient onError(Consumer<JsonNode> callback) {
 
     @Override
     public EventSourceClient onException(Consumer<Throwable> callback) {
-        this.onFailureCallback = callback;
+        this.onExceptionCallback = callback;
         return this;
     }
 
@@ -166,8 +166,9 @@ public EventSourceClient open() {
                     .register(SseFeature.class)
                     .register((Feature) context -> {
                         context.register((ClientRequestFilter) requestContext -> {
-                            requestContext.getHeaders().get("Accept").clear();
-                            requestContext.getHeaders().get("Accept").add(this.incrementalCache ? SseFeature.SERVER_SENT_EVENTS : MediaType.APPLICATION_JSON);
+                            if (!this.incrementalCache) {
+                                requestContext.getHeaders().get("Accept").add(MediaType.APPLICATION_JSON);
+                            }
                         });
                         return true;
                     })
@@ -199,7 +200,7 @@ public void onEvent(InboundEvent inboundEvent) {
                                 onDataCallback.accept(data);
                             } catch (IOException e) {
                                 // notify consumer
-                                onFailureCallback.accept(e);
+                                onExceptionCallback.accept(e);
                             }
                             break;
 
@@ -219,7 +220,7 @@ public void onEvent(InboundEvent inboundEvent) {
                                 onPatchCallback.accept(lastPatch);
 
                             } catch (IOException e) {
-                                onFailureCallback.accept(e);
+                                onExceptionCallback.accept(e);
                             }
                             break;
 
@@ -228,11 +229,11 @@ public void onEvent(InboundEvent inboundEvent) {
                                 LOGGER.debug("Receiving error {} ", eventData);
 
                                 // read the error
-                                JsonNode lastPatch = jsonObjectMapper.readTree(eventData);
+                                JsonNode error = jsonObjectMapper.readTree(eventData);
 
-                                onErrorCallback.accept(lastPatch);
+                                onErrorCallback.accept(error);
                             } catch (IOException e) {
-                                onFailureCallback.accept(e);
+                                onExceptionCallback.accept(e);
                             }
                             break;
 
@@ -248,7 +249,7 @@ public void onEvent(InboundEvent inboundEvent) {
             if (this.onOpenCallback != null)
                 this.onOpenCallback.run();
         } catch (Exception e) {
-            onFailureCallback.accept(e);
+            onExceptionCallback.accept(e);
             this.close();
             System.exit(1);
         }

src/main/java/io/streamdata/sdk/impl/RxJavaEventSourceClientImpl.java

@@ -26,8 +26,8 @@ public RxJavaEventSourceClient addHeader(String name, String value) {
     }
 
     @Override
-    public RxJavaEventSourceClient incrementalCache(boolean enableIncrementalCache) {
-        this.eventSourceClient.incrementalCache(enableIncrementalCache);
+    public RxJavaEventSourceClient useJsonPatch(boolean enableIncrementalCache) {
+        this.eventSourceClient.useJsonPatch(enableIncrementalCache);
         return this;
     }