@Override
    public GraphStageLogic createLogic(Attributes attr) throws Exception {
        JsonParser parser = new JsonParser();
        return new GraphStageLogic(shape) {
            {
                setHandler(out, new AbstractOutHandler() {
                    @Override
                    public void onPull() throws Exception {
                        List<JSONEvent> events = new ArrayList<>();
                        parseInto(events);
                        if (events.isEmpty()) {
                            pull(in);
                        } else {
                            emitMultiple(out, events.iterator());
                        }
                    }
                });
                setHandler(in, new AbstractInHandler() {
                    @Override
                    public void onPush() throws Exception {
                        List<JSONEvent> events = new ArrayList<>();
                        ByteString bytes = grab(in);
                        // TODO either PR ByteBuffer support into actson, or sneaky-access the underlying byte array fields here
                        for (ByteBuffer b: bytes.getByteBuffers()) {
                            byte[] buf= new byte[b.remaining()];
                            b.get(buf, 0, b.remaining());
                            int i = 0;
                            while (i < buf.length) {
                              i += parser.getFeeder().feed(buf, i, buf.length - i);
                              parseInto(events);
                            }
                        }
                        if (events.isEmpty()) {
                            pull(in);
                        } else {
                            emitMultiple(out, events.iterator());
                        }
                    }
                    public void onUpstreamFinish() throws Exception {
                        parser.getFeeder().done();
                        List<JSONEvent> events = new ArrayList<>();
                        parseInto(events);
                        emitMultiple(out, events.iterator());
                        complete(out);
                    }
                });
            }
            private void parseInto(List<JSONEvent> events) {
                Option<JSONEvent> evt = next();
                while (evt.isDefined()) {
                    events.add(evt.get());
                    evt = next();
                }
            };
            private Option<JSONEvent> next() {
                switch(parser.nextEvent()) {
                case JsonEvent.END_ARRAY: return some(JSONEvent.END_ARRAY);
                case JsonEvent.END_OBJECT: return some(JSONEvent.END_OBJECT);
                case JsonEvent.ERROR: throw new IllegalArgumentException("There was a parse error at around character " + parser.getParsedCharacterCount());
                case JsonEvent.EOF: return none();
                case JsonEvent.FIELD_NAME: return some(new JSONEvent.FieldName(parser.getCurrentString()));
                case JsonEvent.NEED_MORE_INPUT: return none();
                case JsonEvent.START_ARRAY: return some(JSONEvent.START_ARRAY);
                case JsonEvent.START_OBJECT: return some(JSONEvent.START_OBJECT);
                case JsonEvent.VALUE_DOUBLE: return some(new JSONEvent.NumericValue(String.valueOf(parser.getCurrentDouble())));
                case JsonEvent.VALUE_FALSE: return some(JSONEvent.FALSE);
                case JsonEvent.VALUE_INT: return some(new JSONEvent.NumericValue(parser.getCurrentString()));
                case JsonEvent.VALUE_NULL: return some(JSONEvent.NULL);
                case JsonEvent.VALUE_STRING: return some(new JSONEvent.StringValue(parser.getCurrentString()));
                case JsonEvent.VALUE_TRUE: return some(JSONEvent.TRUE);
                default: throw new UnsupportedOperationException("Unexpected event in JSON parser");
                }
            }
        };
    }
@Override
    public GraphStageLogic createLogic(Attributes attr) throws Exception {
        JsonParser parser = new JsonParser();
        return new GraphStageLogic(shape) {
            {
                setHandler(out, new AbstractOutHandler() {
                    @Override
                    public void onPull() throws Exception {
                        List<JSONEvent> events = new ArrayList<>();
                        parseInto(events);
                        if (events.isEmpty()) {
                            pull(in);
                        } else {
                            emitMultiple(out, events.iterator());
                        }
                    }
                });
                setHandler(in, new AbstractInHandler() {
                    @Override
                    public void onPush() throws Exception {
                        List<JSONEvent> events = new ArrayList<>();
                        ByteString bytes = grab(in);
                       
                        for (ByteBuffer b: bytes.getByteBuffers()) {
                            byte[] buf= new byte[b.remaining()];
                            b.get(buf, 0, b.remaining());
                            int i = 0;
                            while (i < buf.length) {
                              i += parser.getFeeder().feed(buf, i, buf.length - i);
                              parseInto(events);
                            }
                        }
                        if (events.isEmpty()) {
                            pull(in);
                        } else {
                            emitMultiple(out, events.iterator());
                        }
                    }
                    public void onUpstreamFinish() throws Exception {
                        parser.getFeeder().done();
                        List<JSONEvent> events = new ArrayList<>();
                        parseInto(events);
                        emitMultiple(out, events.iterator());
                        complete(out);
                    }
                });
            }
            private void parseInto(List<JSONEvent> events) {
                Option<JSONEvent> evt = next();
                while (evt.isDefined()) {
                    events.add(evt.get());
                    evt = next();
                }
            };
            private Option<JSONEvent> next() {
                switch(parser.nextEvent()) {
                case JsonEvent.END_ARRAY: return some(JSONEvent.END_ARRAY);
                case JsonEvent.END_OBJECT: return some(JSONEvent.END_OBJECT);
                case JsonEvent.ERROR: throw new IllegalArgumentException("There was a parse error at around character " + parser.getParsedCharacterCount());
                case JsonEvent.EOF: return none();
                case JsonEvent.FIELD_NAME: return some(new JSONEvent.FieldName(parser.getCurrentString()));
                case JsonEvent.NEED_MORE_INPUT: return none();
                case JsonEvent.START_ARRAY: return some(JSONEvent.START_ARRAY);
                case JsonEvent.START_OBJECT: return some(JSONEvent.START_OBJECT);
                case JsonEvent.VALUE_DOUBLE: return some(new JSONEvent.NumericValue(String.valueOf(parser.getCurrentDouble())));
                case JsonEvent.VALUE_FALSE: return some(JSONEvent.FALSE);
                case JsonEvent.VALUE_INT: return some(new JSONEvent.NumericValue(parser.getCurrentString()));
                case JsonEvent.VALUE_NULL: return some(JSONEvent.NULL);
                case JsonEvent.VALUE_STRING: return some(new JSONEvent.StringValue(parser.getCurrentString()));
                case JsonEvent.VALUE_TRUE: return some(JSONEvent.TRUE);
                default: throw new UnsupportedOperationException("Unexpected event in JSON parser");
                }
            }
        };
    }
@override public graphstagelogic createlogic(attributes attr) throws exception { jsonparser parser = new jsonparser(); return new graphstagelogic(shape) { { sethandler(out, new abstractouthandler() { @override public void onpull() throws exception { list<jsonevent> events = new arraylist<>(); parseinto(events); if (events.isempty()) { pull(in); } else { emitmultiple(out, events.iterator()); } } }); sethandler(in, new abstractinhandler() { @override public void onpush() throws exception { list<jsonevent> events = new arraylist<>(); bytestring bytes = grab(in); for (bytebuffer b: bytes.getbytebuffers()) { byte[] buf= new byte[b.remaining()]; b.get(buf, 0, b.remaining()); int i = 0; while (i < buf.length) { i += parser.getfeeder().feed(buf, i, buf.length - i); parseinto(events); } } if (events.isempty()) { pull(in); } else { emitmultiple(out, events.iterator()); } } public void onupstreamfinish() throws exception { parser.getfeeder().done(); list<jsonevent> events = new arraylist<>(); parseinto(events); emitmultiple(out, events.iterator()); complete(out); } }); } private void parseinto(list<jsonevent> events) { option<jsonevent> evt = next(); while (evt.isdefined()) { events.add(evt.get()); evt = next(); } }; private option<jsonevent> next() { switch(parser.nextevent()) { case jsonevent.end_array: return some(jsonevent.end_array); case jsonevent.end_object: return some(jsonevent.end_object); case jsonevent.error: throw new illegalargumentexception("there was a parse error at around character " + parser.getparsedcharactercount()); case jsonevent.eof: return none(); case jsonevent.field_name: return some(new jsonevent.fieldname(parser.getcurrentstring())); case jsonevent.need_more_input: return none(); case jsonevent.start_array: return some(jsonevent.start_array); case jsonevent.start_object: return some(jsonevent.start_object); case jsonevent.value_double: return some(new jsonevent.numericvalue(string.valueof(parser.getcurrentdouble()))); case jsonevent.value_false: return some(jsonevent.false); case jsonevent.value_int: return some(new jsonevent.numericvalue(parser.getcurrentstring())); case jsonevent.value_null: return some(jsonevent.null); case jsonevent.value_string: return some(new jsonevent.stringvalue(parser.getcurrentstring())); case jsonevent.value_true: return some(jsonevent.true); default: throw new unsupportedoperationexception("unexpected event in json parser"); } } }; }