@Test void lifecycle() { final Component delegate = new Component(); final Input input = new InputImpl("Root", "Test", "Plugin", delegate); assertFalse(delegate.start); assertFalse(delegate.stop); assertEquals(0, delegate.count); input.start(); assertTrue(delegate.start); assertFalse(delegate.stop); assertEquals(0, delegate.count); IntStream.range(0, 10).forEach(i -> { final Object next = input.next(); assertTrue(Record.class.isInstance(next)); final Record record = Record.class.cast(next); assertEquals(Schema.Type.RECORD, record.getSchema().getType()); assertEquals(1, record.getSchema().getEntries().size()); final Schema.Entry data = record.getSchema().getEntries().iterator().next(); assertEquals("data", data.getName()); assertEquals(Schema.Type.DOUBLE, data.getType()); assertEquals(i, record.get(Double.class, "data").doubleValue()); assertTrue(delegate.start); assertFalse(delegate.stop); assertEquals(i + 1, delegate.count); }); input.stop(); assertTrue(delegate.start); assertTrue(delegate.stop); assertEquals(10, delegate.count); }
@Override public Object next() { if (next == null) { init(); } final Object record = readNext(); if (record == null) { return null; } final Class<?> recordClass = record.getClass(); if (recordClass.isPrimitive() || String.class == recordClass) { return record; } return converters.toRecord(registry, record, this::jsonb, this::recordBuilderFactory); }
InputImpl extends LifecycleImpl implements Input, Delegated { @Override public Object next() { if (next == null) { init(); } final Object record = readNext(); if (record == null) { return null; } final Class<?> recordClass = record.getClass(); if (recordClass.isPrimitive() || String.class == recordClass) { return record; } return converters.toRecord(registry, record, this::jsonb, this::recordBuilderFactory); } }
InputImpl extends LifecycleImpl implements Input, Delegated { @Override public Object next() { if (next == null) { init(); } final Object record = readNext(); if (record == null) { return null; } final Class<?> recordClass = record.getClass(); if (recordClass.isPrimitive() || String.class == recordClass) { return record; } return converters.toRecord(registry, record, this::jsonb, this::recordBuilderFactory); } InputImpl(final String rootName, final String name, final String plugin, final Serializable instance); protected  InputImpl(); }
InputImpl extends LifecycleImpl implements Input, Delegated { @Override public Object next() { if (next == null) { init(); } final Object record = readNext(); if (record == null) { return null; } final Class<?> recordClass = record.getClass(); if (recordClass.isPrimitive() || String.class == recordClass) { return record; } return converters.toRecord(registry, record, this::jsonb, this::recordBuilderFactory); } InputImpl(final String rootName, final String name, final String plugin, final Serializable instance); protected  InputImpl(); @Override Object next(); @Override Object getDelegate(); }
InputImpl extends LifecycleImpl implements Input, Delegated { @Override public Object next() { if (next == null) { init(); } final Object record = readNext(); if (record == null) { return null; } final Class<?> recordClass = record.getClass(); if (recordClass.isPrimitive() || String.class == recordClass) { return record; } return converters.toRecord(registry, record, this::jsonb, this::recordBuilderFactory); } InputImpl(final String rootName, final String name, final String plugin, final Serializable instance); protected  InputImpl(); @Override Object next(); @Override Object getDelegate();  }