class Kitap implements Runnable {

    private void readAll(final JsonObject config, final JsonObject stream, final String streamName,
            final Mapper mapper) {
        final AtomicReference<Schema> lastSchema = new AtomicReference<>();
        records(mapper).peek(record -> {
            final Schema newSchema = record.getSchema();
            if (!newSchema.equals(lastSchema.get())) {
                final JsonArray keys = config != null && config.containsKey("keys") ? config.getJsonArray("keys")
                        : jsonBuilderFactory.createArrayBuilder().build();
                final JsonArray bookmarks =
                        config != null && config.containsKey("bookmarks") ? config.getJsonArray("bookmarks")
                                : jsonBuilderFactory.createArrayBuilder().build();
                final JsonObject schema = ofNullable(stream.getJsonObject("schema"))
                        .orElseGet(() -> new JsonSchemaGenerator(newSchema.getEntries(), jsonBuilderFactory).get());
                singer.writeSchema(streamName, schema, keys, bookmarks);
                lastSchema.set(newSchema);
            }
        }).forEach(record -> singer.writeRecord(streamName, recordJsonMapper.apply(record)));
    }

    public  Kitap(final String... args);
    public  Kitap(final SingerArgs args, final Singer singer);

}