// MIT License
//
// Copyright (c) 2016 Michel Kraemer
//
// Permission is hereby granted, free of charge, to any person obtaining
// a copy of this software and associated documentation files (the
// "Software"), to deal in the Software without restriction, including
// without limitation the rights to use, copy, modify, merge, publish,
// distribute, sublicense, and/or sell copies of the Software, and to
// permit persons to whom the Software is furnished to do so, subject to
// the following conditions:
//
// The above copyright notice and this permission notice shall be
// included in all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
// NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
// LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

package rxjava;

import de.undercouch.actson.JsonEvent;
import de.undercouch.actson.JsonParser;
import rx.Observable.Operator;
import rx.Subscriber;

/**
 * A reusable operator function that you can lift into an RxJava
 * {@link rx.Observable}. It transforms byte arrays into JSON events.
 * @author Michel Kraemer
 */
public class JsonParserOperator implements Operator<Result, byte[]> {
  /**
   * Non-blocking JSON parser
   */
  private JsonParser parser = new JsonParser();
  
  /**
   * Process events from the parser until it needs more input. Notify the
   * subscriber accordingly.
   * @param s the subscriber
   */
  private void processEvents(Subscriber<? super Result> s) {
    int event;
    do {
      event = parser.nextEvent();
      
      // handle event and notify subscriber
      if (event == JsonEvent.VALUE_STRING) {
        // forward string values to subscriber
        s.onNext(new Result(event, parser.getCurrentString()));
      } else if (event == JsonEvent.EOF) {
        // notify the subscriber that the observable has finished
        s.onNext(new Result(event));
        s.onCompleted();
        break;
      } else if (event == JsonEvent.ERROR) {
        // notify the subscriber about the error
        s.onError(new IllegalStateException("Syntax error"));
        break;
      } else if (event != JsonEvent.NEED_MORE_INPUT) {
        // forward JSON event
        s.onNext(new Result(event));
      }
    } while (event != JsonEvent.NEED_MORE_INPUT);
  }
  
  @Override
  public Subscriber<? super byte[]> call(Subscriber<? super Result> s) {
    return new Subscriber<byte[]>(s) {
      @Override
      public void onCompleted() {
        if (!s.isUnsubscribed()) {
          // finish parsing and forward events to the
          // subscriber (including the EOF event)
          parser.getFeeder().done();
          processEvents(s);
        }
      }

      @Override
      public void onError(Throwable e) {
        if (!s.isUnsubscribed()) {
          s.onError(e);
        }
      }

      @Override
      public void onNext(byte[] buf) {
        if (s.isUnsubscribed()) {
          return;
        }
        
        // push bytes into the parser and then process JSON events
        int i = 0;
        while (i < buf.length) {
          i += parser.getFeeder().feed(buf, i, buf.length - i);
          processEvents(s);
        }
      }
    };
  }
}

@@ -1747,16 +1747,154 @@
 scriber%0A
+   * @return true if the caller should continue parsing, false if there was an%0A   * error or if the end of the JSON text has been reached%0A
    */%0A  
@@ -1901,20 +1901,23 @@
 private 
-void
+boolean
  process
@@ -2391,37 +2391,44 @@
 eted();%0A        
-b
 re
-ak
+turn false
 ;%0A      %7D else i
@@ -2581,13 +2581,20 @@
     
-b
 re
-ak
+turn false
 ;%0A  
@@ -2771,16 +2771,38 @@
 INPUT);%0A
+    %0A    return true;%0A
   %7D%0A  %0A 
@@ -3643,32 +3643,37 @@
 - i);%0A          
+if (!
 processEvents(s)
@@ -3664,33 +3664,66 @@
 processEvents(s)
-;
+) %7B%0A            break;%0A          %7D
 %0A        %7D%0A     

Stop parsing on error in RxJava example