@Test void visit() { final Collection<String> visited = new ArrayList<>(); final AtomicInteger out = new AtomicInteger(); assertEquals(3, service .visit(RecordVisitor.class .cast(Proxy .newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[] { RecordVisitor.class }, (proxy, method, args) -> { visited .add(method.getName() + "/" + (args == null ? "null" : Stream .of(args) .filter(it -> !Schema.Entry.class .isInstance(it)) .collect(Collectors.toList()))); switch (method.getName()) { case "get": return out.incrementAndGet(); case "apply": return asList(args) .stream() .mapToInt(Integer.class::cast) .sum(); default: return method.getReturnType() == RecordVisitor.class ? proxy : null; } })), baseRecord)); assertEquals(asList("onString/[Optional[Test]]", "onInt/[OptionalInt[33]]", "onRecord/[Optional[{\"street\":\"here\",\"number\":1}]]", "onString/[Optional[here]]", "onInt/[OptionalInt[1]]", "get/null", "get/null", "apply/[1, 2]"), visited); }
@Override public <T> T visit(final RecordVisitor<T> visitor, final Record record) { final AtomicReference<T> out = new AtomicReference<>(); record.getSchema().getEntries().forEach(entry -> { switch (entry.getType()) { case INT: visitor.onInt(entry, record.getOptionalInt(entry.getName())); break; case LONG: visitor.onLong(entry, record.getOptionalLong(entry.getName())); break; case FLOAT: visitor.onFloat(entry, record.getOptionalFloat(entry.getName())); break; case DOUBLE: visitor.onDouble(entry, record.getOptionalDouble(entry.getName())); break; case BOOLEAN: visitor.onBoolean(entry, record.getOptionalBoolean(entry.getName())); break; case STRING: visitor.onString(entry, record.getOptionalString(entry.getName())); break; case DATETIME: visitor.onDatetime(entry, record.getOptionalDateTime(entry.getName())); break; case BYTES: visitor.onBytes(entry, record.getOptionalBytes(entry.getName())); break; case RECORD: final Optional<Record> optionalRecord = record.getOptionalRecord(entry.getName()); final RecordVisitor<T> recordVisitor = visitor.onRecord(entry, optionalRecord); optionalRecord.ifPresent(r -> { final T visited = visit(recordVisitor, r); if (visited != null) { final T current = out.get(); out.set(current == null ? visited : visitor.apply(current, visited)); } }); break; case ARRAY: final Schema schema = entry.getElementSchema(); switch (schema.getType()) { case INT: visitor.onIntArray(entry, record.getOptionalArray(int.class, entry.getName())); break; case LONG: visitor.onLongArray(entry, record.getOptionalArray(long.class, entry.getName())); break; case FLOAT: visitor.onFloatArray(entry, record.getOptionalArray(float.class, entry.getName())); break; case DOUBLE: visitor.onDoubleArray(entry, record.getOptionalArray(double.class, entry.getName())); break; case BOOLEAN: visitor.onBooleanArray(entry, record.getOptionalArray(boolean.class, entry.getName())); break; case STRING: visitor.onStringArray(entry, record.getOptionalArray(String.class, entry.getName())); break; case DATETIME: visitor.onDatetimeArray(entry, record.getOptionalArray(ZonedDateTime.class, entry.getName())); break; case BYTES: visitor.onBytesArray(entry, record.getOptionalArray(byte[].class, entry.getName())); break; case RECORD: final Optional<Collection<Record>> array = record.getOptionalArray(Record.class, entry.getName()); final RecordVisitor<T> recordArrayVisitor = visitor.onRecordArray(entry, array); array.ifPresent(a -> a.forEach(r -> { final T visited = visit(recordArrayVisitor, r); if (visited != null) { final T current = out.get(); out.set(current == null ? visited : visitor.apply(current, visited)); } })); break; default: throw new IllegalStateException("Unsupported entry type: " + entry); } break; default: throw new IllegalStateException("Unsupported entry type: " + entry); } }); final T value = out.get(); final T visited = visitor.get(); if (value != null) { return visitor.apply(value, visited); } return visited; }
RecordServiceImpl implements RecordService, Serializable { @Override public <T> T visit(final RecordVisitor<T> visitor, final Record record) { final AtomicReference<T> out = new AtomicReference<>(); record.getSchema().getEntries().forEach(entry -> { switch (entry.getType()) { case INT: visitor.onInt(entry, record.getOptionalInt(entry.getName())); break; case LONG: visitor.onLong(entry, record.getOptionalLong(entry.getName())); break; case FLOAT: visitor.onFloat(entry, record.getOptionalFloat(entry.getName())); break; case DOUBLE: visitor.onDouble(entry, record.getOptionalDouble(entry.getName())); break; case BOOLEAN: visitor.onBoolean(entry, record.getOptionalBoolean(entry.getName())); break; case STRING: visitor.onString(entry, record.getOptionalString(entry.getName())); break; case DATETIME: visitor.onDatetime(entry, record.getOptionalDateTime(entry.getName())); break; case BYTES: visitor.onBytes(entry, record.getOptionalBytes(entry.getName())); break; case RECORD: final Optional<Record> optionalRecord = record.getOptionalRecord(entry.getName()); final RecordVisitor<T> recordVisitor = visitor.onRecord(entry, optionalRecord); optionalRecord.ifPresent(r -> { final T visited = visit(recordVisitor, r); if (visited != null) { final T current = out.get(); out.set(current == null ? visited : visitor.apply(current, visited)); } }); break; case ARRAY: final Schema schema = entry.getElementSchema(); switch (schema.getType()) { case INT: visitor.onIntArray(entry, record.getOptionalArray(int.class, entry.getName())); break; case LONG: visitor.onLongArray(entry, record.getOptionalArray(long.class, entry.getName())); break; case FLOAT: visitor.onFloatArray(entry, record.getOptionalArray(float.class, entry.getName())); break; case DOUBLE: visitor.onDoubleArray(entry, record.getOptionalArray(double.class, entry.getName())); break; case BOOLEAN: visitor.onBooleanArray(entry, record.getOptionalArray(boolean.class, entry.getName())); break; case STRING: visitor.onStringArray(entry, record.getOptionalArray(String.class, entry.getName())); break; case DATETIME: visitor.onDatetimeArray(entry, record.getOptionalArray(ZonedDateTime.class, entry.getName())); break; case BYTES: visitor.onBytesArray(entry, record.getOptionalArray(byte[].class, entry.getName())); break; case RECORD: final Optional<Collection<Record>> array = record.getOptionalArray(Record.class, entry.getName()); final RecordVisitor<T> recordArrayVisitor = visitor.onRecordArray(entry, array); array.ifPresent(a -> a.forEach(r -> { final T visited = visit(recordArrayVisitor, r); if (visited != null) { final T current = out.get(); out.set(current == null ? visited : visitor.apply(current, visited)); } })); break; default: throw new IllegalStateException("Unsupported entry type: " + entry); } break; default: throw new IllegalStateException("Unsupported entry type: " + entry); } }); final T value = out.get(); final T visited = visitor.get(); if (value != null) { return visitor.apply(value, visited); } return visited; } }
RecordServiceImpl implements RecordService, Serializable { @Override public <T> T visit(final RecordVisitor<T> visitor, final Record record) { final AtomicReference<T> out = new AtomicReference<>(); record.getSchema().getEntries().forEach(entry -> { switch (entry.getType()) { case INT: visitor.onInt(entry, record.getOptionalInt(entry.getName())); break; case LONG: visitor.onLong(entry, record.getOptionalLong(entry.getName())); break; case FLOAT: visitor.onFloat(entry, record.getOptionalFloat(entry.getName())); break; case DOUBLE: visitor.onDouble(entry, record.getOptionalDouble(entry.getName())); break; case BOOLEAN: visitor.onBoolean(entry, record.getOptionalBoolean(entry.getName())); break; case STRING: visitor.onString(entry, record.getOptionalString(entry.getName())); break; case DATETIME: visitor.onDatetime(entry, record.getOptionalDateTime(entry.getName())); break; case BYTES: visitor.onBytes(entry, record.getOptionalBytes(entry.getName())); break; case RECORD: final Optional<Record> optionalRecord = record.getOptionalRecord(entry.getName()); final RecordVisitor<T> recordVisitor = visitor.onRecord(entry, optionalRecord); optionalRecord.ifPresent(r -> { final T visited = visit(recordVisitor, r); if (visited != null) { final T current = out.get(); out.set(current == null ? visited : visitor.apply(current, visited)); } }); break; case ARRAY: final Schema schema = entry.getElementSchema(); switch (schema.getType()) { case INT: visitor.onIntArray(entry, record.getOptionalArray(int.class, entry.getName())); break; case LONG: visitor.onLongArray(entry, record.getOptionalArray(long.class, entry.getName())); break; case FLOAT: visitor.onFloatArray(entry, record.getOptionalArray(float.class, entry.getName())); break; case DOUBLE: visitor.onDoubleArray(entry, record.getOptionalArray(double.class, entry.getName())); break; case BOOLEAN: visitor.onBooleanArray(entry, record.getOptionalArray(boolean.class, entry.getName())); break; case STRING: visitor.onStringArray(entry, record.getOptionalArray(String.class, entry.getName())); break; case DATETIME: visitor.onDatetimeArray(entry, record.getOptionalArray(ZonedDateTime.class, entry.getName())); break; case BYTES: visitor.onBytesArray(entry, record.getOptionalArray(byte[].class, entry.getName())); break; case RECORD: final Optional<Collection<Record>> array = record.getOptionalArray(Record.class, entry.getName()); final RecordVisitor<T> recordArrayVisitor = visitor.onRecordArray(entry, array); array.ifPresent(a -> a.forEach(r -> { final T visited = visit(recordArrayVisitor, r); if (visited != null) { final T current = out.get(); out.set(current == null ? visited : visitor.apply(current, visited)); } })); break; default: throw new IllegalStateException("Unsupported entry type: " + entry); } break; default: throw new IllegalStateException("Unsupported entry type: " + entry); } }); final T value = out.get(); final T visited = visitor.get(); if (value != null) { return visitor.apply(value, visited); } return visited; }  }
RecordServiceImpl implements RecordService, Serializable { @Override public <T> T visit(final RecordVisitor<T> visitor, final Record record) { final AtomicReference<T> out = new AtomicReference<>(); record.getSchema().getEntries().forEach(entry -> { switch (entry.getType()) { case INT: visitor.onInt(entry, record.getOptionalInt(entry.getName())); break; case LONG: visitor.onLong(entry, record.getOptionalLong(entry.getName())); break; case FLOAT: visitor.onFloat(entry, record.getOptionalFloat(entry.getName())); break; case DOUBLE: visitor.onDouble(entry, record.getOptionalDouble(entry.getName())); break; case BOOLEAN: visitor.onBoolean(entry, record.getOptionalBoolean(entry.getName())); break; case STRING: visitor.onString(entry, record.getOptionalString(entry.getName())); break; case DATETIME: visitor.onDatetime(entry, record.getOptionalDateTime(entry.getName())); break; case BYTES: visitor.onBytes(entry, record.getOptionalBytes(entry.getName())); break; case RECORD: final Optional<Record> optionalRecord = record.getOptionalRecord(entry.getName()); final RecordVisitor<T> recordVisitor = visitor.onRecord(entry, optionalRecord); optionalRecord.ifPresent(r -> { final T visited = visit(recordVisitor, r); if (visited != null) { final T current = out.get(); out.set(current == null ? visited : visitor.apply(current, visited)); } }); break; case ARRAY: final Schema schema = entry.getElementSchema(); switch (schema.getType()) { case INT: visitor.onIntArray(entry, record.getOptionalArray(int.class, entry.getName())); break; case LONG: visitor.onLongArray(entry, record.getOptionalArray(long.class, entry.getName())); break; case FLOAT: visitor.onFloatArray(entry, record.getOptionalArray(float.class, entry.getName())); break; case DOUBLE: visitor.onDoubleArray(entry, record.getOptionalArray(double.class, entry.getName())); break; case BOOLEAN: visitor.onBooleanArray(entry, record.getOptionalArray(boolean.class, entry.getName())); break; case STRING: visitor.onStringArray(entry, record.getOptionalArray(String.class, entry.getName())); break; case DATETIME: visitor.onDatetimeArray(entry, record.getOptionalArray(ZonedDateTime.class, entry.getName())); break; case BYTES: visitor.onBytesArray(entry, record.getOptionalArray(byte[].class, entry.getName())); break; case RECORD: final Optional<Collection<Record>> array = record.getOptionalArray(Record.class, entry.getName()); final RecordVisitor<T> recordArrayVisitor = visitor.onRecordArray(entry, array); array.ifPresent(a -> a.forEach(r -> { final T visited = visit(recordArrayVisitor, r); if (visited != null) { final T current = out.get(); out.set(current == null ? visited : visitor.apply(current, visited)); } })); break; default: throw new IllegalStateException("Unsupported entry type: " + entry); } break; default: throw new IllegalStateException("Unsupported entry type: " + entry); } }); final T value = out.get(); final T visited = visitor.get(); if (value != null) { return visitor.apply(value, visited); } return visited; }  @Override Collector<Schema.Entry, Record.Builder, Record> toRecord(final Schema schema, final Record fallbackRecord,
            final BiFunction<Schema.Entry, Record.Builder, Boolean> customHandler,
            final BiConsumer<Record.Builder, Boolean> beforeFinish); @Override Record create(final Schema schema, final Record fallbackRecord,
            final BiFunction<Schema.Entry, Record.Builder, Boolean> customHandler,
            final BiConsumer<Record.Builder, Boolean> beforeFinish); @Override T visit(final RecordVisitor<T> visitor, final Record record); @Override T toObject(final Record data, final Class<T> expected); @Override Record toRecord(final T data); @Override boolean forwardEntry(final Record source, final Record.Builder builder, final String sourceColumn,
            final Schema.Entry entry); }
RecordServiceImpl implements RecordService, Serializable { @Override public <T> T visit(final RecordVisitor<T> visitor, final Record record) { final AtomicReference<T> out = new AtomicReference<>(); record.getSchema().getEntries().forEach(entry -> { switch (entry.getType()) { case INT: visitor.onInt(entry, record.getOptionalInt(entry.getName())); break; case LONG: visitor.onLong(entry, record.getOptionalLong(entry.getName())); break; case FLOAT: visitor.onFloat(entry, record.getOptionalFloat(entry.getName())); break; case DOUBLE: visitor.onDouble(entry, record.getOptionalDouble(entry.getName())); break; case BOOLEAN: visitor.onBoolean(entry, record.getOptionalBoolean(entry.getName())); break; case STRING: visitor.onString(entry, record.getOptionalString(entry.getName())); break; case DATETIME: visitor.onDatetime(entry, record.getOptionalDateTime(entry.getName())); break; case BYTES: visitor.onBytes(entry, record.getOptionalBytes(entry.getName())); break; case RECORD: final Optional<Record> optionalRecord = record.getOptionalRecord(entry.getName()); final RecordVisitor<T> recordVisitor = visitor.onRecord(entry, optionalRecord); optionalRecord.ifPresent(r -> { final T visited = visit(recordVisitor, r); if (visited != null) { final T current = out.get(); out.set(current == null ? visited : visitor.apply(current, visited)); } }); break; case ARRAY: final Schema schema = entry.getElementSchema(); switch (schema.getType()) { case INT: visitor.onIntArray(entry, record.getOptionalArray(int.class, entry.getName())); break; case LONG: visitor.onLongArray(entry, record.getOptionalArray(long.class, entry.getName())); break; case FLOAT: visitor.onFloatArray(entry, record.getOptionalArray(float.class, entry.getName())); break; case DOUBLE: visitor.onDoubleArray(entry, record.getOptionalArray(double.class, entry.getName())); break; case BOOLEAN: visitor.onBooleanArray(entry, record.getOptionalArray(boolean.class, entry.getName())); break; case STRING: visitor.onStringArray(entry, record.getOptionalArray(String.class, entry.getName())); break; case DATETIME: visitor.onDatetimeArray(entry, record.getOptionalArray(ZonedDateTime.class, entry.getName())); break; case BYTES: visitor.onBytesArray(entry, record.getOptionalArray(byte[].class, entry.getName())); break; case RECORD: final Optional<Collection<Record>> array = record.getOptionalArray(Record.class, entry.getName()); final RecordVisitor<T> recordArrayVisitor = visitor.onRecordArray(entry, array); array.ifPresent(a -> a.forEach(r -> { final T visited = visit(recordArrayVisitor, r); if (visited != null) { final T current = out.get(); out.set(current == null ? visited : visitor.apply(current, visited)); } })); break; default: throw new IllegalStateException("Unsupported entry type: " + entry); } break; default: throw new IllegalStateException("Unsupported entry type: " + entry); } }); final T value = out.get(); final T visited = visitor.get(); if (value != null) { return visitor.apply(value, visited); } return visited; }  @Override Collector<Schema.Entry, Record.Builder, Record> toRecord(final Schema schema, final Record fallbackRecord,
            final BiFunction<Schema.Entry, Record.Builder, Boolean> customHandler,
            final BiConsumer<Record.Builder, Boolean> beforeFinish); @Override Record create(final Schema schema, final Record fallbackRecord,
            final BiFunction<Schema.Entry, Record.Builder, Boolean> customHandler,
            final BiConsumer<Record.Builder, Boolean> beforeFinish); @Override T visit(final RecordVisitor<T> visitor, final Record record); @Override T toObject(final Record data, final Class<T> expected); @Override Record toRecord(final T data); @Override boolean forwardEntry(final Record source, final Record.Builder builder, final String sourceColumn,
            final Schema.Entry entry);  }