// TODO either PR ByteBuffer support into actson, or sneaky-access the underlying byte array fields here
@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");
                }
            }
        };
    }
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()];
} 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()) {
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());