@Test void readAll() throws IOException { final Kitap kitap = new Kitap(new SingerArgs("--config", config.toAbsolutePath().toString()), new Singer(new IO(), () -> CONSTANT_DATE)); kitap.run(); flushIO.run(); try (final BufferedReader reader = new BufferedReader(new StringReader(stdout.toString("UTF-8")))) { final List<String> actuals = reader.lines().collect(toList()); assertLinesMatch(asList( "{\"type\":\"SCHEMA\",\"stream\":\"default\",\"schema\":{\"type\":[\"null\",\"object\"],\"additionalProperties\":false,\"properties\":{\"record_number\":{\"type\":[\"integer\"]}}},\"key_properties\":[],\"bookmark_properties\":[]}", "{\"type\":\"RECORD\",\"stream\":\"default\",\"time_extracted\":\"2019-08-23T15:11:00.000Z\",\"record\":{\"record_number\":1}}", "{\"type\":\"RECORD\",\"stream\":\"default\",\"time_extracted\":\"2019-08-23T15:11:00.000Z\",\"record\":{\"record_number\":2}}", "{\"type\":\"RECORD\",\"stream\":\"default\",\"time_extracted\":\"2019-08-23T15:11:00.000Z\",\"record\":{\"record_number\":3}}", "{\"type\":\"RECORD\",\"stream\":\"default\",\"time_extracted\":\"2019-08-23T15:11:00.000Z\",\"record\":{\"record_number\":4}}", "{\"type\":\"RECORD\",\"stream\":\"default\",\"time_extracted\":\"2019-08-23T15:11:00.000Z\",\"record\":{\"record_number\":5}}", "{\"type\":\"RECORD\",\"stream\":\"default\",\"time_extracted\":\"2019-08-23T15:11:00.000Z\",\"record\":{\"record_number\":6}}", "{\"type\":\"RECORD\",\"stream\":\"default\",\"time_extracted\":\"2019-08-23T15:11:00.000Z\",\"record\":{\"record_number\":7}}", "{\"type\":\"RECORD\",\"stream\":\"default\",\"time_extracted\":\"2019-08-23T15:11:00.000Z\",\"record\":{\"record_number\":8}}", "{\"type\":\"RECORD\",\"stream\":\"default\",\"time_extracted\":\"2019-08-23T15:11:00.000Z\",\"record\":{\"record_number\":9}}", "{\"type\":\"RECORD\",\"stream\":\"default\",\"time_extracted\":\"2019-08-23T15:11:00.000Z\",\"record\":{\"record_number\":10}}"), actuals, actuals::toString); } try (final BufferedReader reader = new BufferedReader(new StringReader(stderr.toString("UTF-8")))) { final String dateRegex = "\\p{Digit}{4}-\\p{Digit}{2}-\\p{Digit}{2}.\\p{Digit}{2}:\\p{Digit}{2}:\\p{Digit}{2}.\\p{Digit}{1,3}"; final List<String> actuals = reader.lines().collect(toList()); assertLinesMatch( asList("\\[" + dateRegex + "\\]\\[main\\]\\[ERROR\\]\\[log4j2\\] log4j2 error", "\\[" + dateRegex + "\\]\\[main\\]\\[ERROR\\]\\[log4j\\] log4j error", "\\[" + dateRegex + "\\]\\[main\\]\\[ERROR\\]\\[logback\\] logback error", "\\[" + dateRegex + "Z\\]\\[main\\]\\[SEVERE\\]\\[jul\\] jul error"), actuals, actuals::toString); } }
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))); }
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))); } }
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))); } Kitap(final String... args);  Kitap(final SingerArgs args, final Singer singer); }
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))); } Kitap(final String... args);  Kitap(final SingerArgs args, final Singer singer); @Override void run(); static void main(final String... args); }
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))); } Kitap(final String... args);  Kitap(final SingerArgs args, final Singer singer); @Override void run(); static void main(final String... args);  }