@Test void testOnBlockingQueue() { ArrayBlockingQueue<Plc4xSchema.Record> queue = new ArrayBlockingQueue<>(100); Plc4xStreamTable table = new Plc4xStreamTable(queue, new JobConfigurationImpl( "job1", null, 100, Collections.emptyList(), Collections.singletonMap("key", "address"))); Map<String, Object> objects = Collections.singletonMap("key", "value"); queue.add(new Plc4xSchema.Record(Instant.now(), "", objects)); Enumerator<Object[]> enumerator = table.scan(null).enumerator(); assertThat(enumerator.moveNext()).isTrue(); assertThat(enumerator.current()).contains("value"); }
public Enumerable<Object[]> scan(DataContext root) { return new AbstractEnumerable<Object[]>() { @Override public Enumerator<Object[]> enumerator() { return new Enumerator<Object[]>() { private final AtomicLong counter = new AtomicLong(0); @Override public Object[] current() { List<Object> objects = new ArrayList<>(Arrays.asList(new Timestamp(current.timestamp.toEpochMilli()), current.source)); List<Object> objects2 = names.stream().map(name -> current.values.get(name)).collect(Collectors.toList()); objects.addAll(objects2); return objects.toArray(); } @Override public boolean moveNext() { try { current = queue.take(); if (tableCutoff <= 0L) { return true; } return counter.getAndIncrement() < tableCutoff; } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return false; } @Override public void reset() { counter.set(0); } @Override public void close() { } }; } }; }
Plc4xBaseTable extends AbstractTable { public Enumerable<Object[]> scan(DataContext root) { return new AbstractEnumerable<Object[]>() { @Override public Enumerator<Object[]> enumerator() { return new Enumerator<Object[]>() { private final AtomicLong counter = new AtomicLong(0); @Override public Object[] current() { List<Object> objects = new ArrayList<>(Arrays.asList(new Timestamp(current.timestamp.toEpochMilli()), current.source)); List<Object> objects2 = names.stream().map(name -> current.values.get(name)).collect(Collectors.toList()); objects.addAll(objects2); return objects.toArray(); } @Override public boolean moveNext() { try { current = queue.take(); if (tableCutoff <= 0L) { return true; } return counter.getAndIncrement() < tableCutoff; } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return false; } @Override public void reset() { counter.set(0); } @Override public void close() { } }; } }; } }
Plc4xBaseTable extends AbstractTable { public Enumerable<Object[]> scan(DataContext root) { return new AbstractEnumerable<Object[]>() { @Override public Enumerator<Object[]> enumerator() { return new Enumerator<Object[]>() { private final AtomicLong counter = new AtomicLong(0); @Override public Object[] current() { List<Object> objects = new ArrayList<>(Arrays.asList(new Timestamp(current.timestamp.toEpochMilli()), current.source)); List<Object> objects2 = names.stream().map(name -> current.values.get(name)).collect(Collectors.toList()); objects.addAll(objects2); return objects.toArray(); } @Override public boolean moveNext() { try { current = queue.take(); if (tableCutoff <= 0L) { return true; } return counter.getAndIncrement() < tableCutoff; } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return false; } @Override public void reset() { counter.set(0); } @Override public void close() { } }; } }; } Plc4xBaseTable(BlockingQueue<Plc4xSchema.Record> queue, JobConfiguration conf, long tableCutoff); }
Plc4xBaseTable extends AbstractTable { public Enumerable<Object[]> scan(DataContext root) { return new AbstractEnumerable<Object[]>() { @Override public Enumerator<Object[]> enumerator() { return new Enumerator<Object[]>() { private final AtomicLong counter = new AtomicLong(0); @Override public Object[] current() { List<Object> objects = new ArrayList<>(Arrays.asList(new Timestamp(current.timestamp.toEpochMilli()), current.source)); List<Object> objects2 = names.stream().map(name -> current.values.get(name)).collect(Collectors.toList()); objects.addAll(objects2); return objects.toArray(); } @Override public boolean moveNext() { try { current = queue.take(); if (tableCutoff <= 0L) { return true; } return counter.getAndIncrement() < tableCutoff; } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return false; } @Override public void reset() { counter.set(0); } @Override public void close() { } }; } }; } Plc4xBaseTable(BlockingQueue<Plc4xSchema.Record> queue, JobConfiguration conf, long tableCutoff); @Override Statistic getStatistic(); @Override RelDataType getRowType(RelDataTypeFactory typeFactory); Enumerable<Object[]> scan(DataContext root); }
Plc4xBaseTable extends AbstractTable { public Enumerable<Object[]> scan(DataContext root) { return new AbstractEnumerable<Object[]>() { @Override public Enumerator<Object[]> enumerator() { return new Enumerator<Object[]>() { private final AtomicLong counter = new AtomicLong(0); @Override public Object[] current() { List<Object> objects = new ArrayList<>(Arrays.asList(new Timestamp(current.timestamp.toEpochMilli()), current.source)); List<Object> objects2 = names.stream().map(name -> current.values.get(name)).collect(Collectors.toList()); objects.addAll(objects2); return objects.toArray(); } @Override public boolean moveNext() { try { current = queue.take(); if (tableCutoff <= 0L) { return true; } return counter.getAndIncrement() < tableCutoff; } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return false; } @Override public void reset() { counter.set(0); } @Override public void close() { } }; } }; } Plc4xBaseTable(BlockingQueue<Plc4xSchema.Record> queue, JobConfiguration conf, long tableCutoff); @Override Statistic getStatistic(); @Override RelDataType getRowType(RelDataTypeFactory typeFactory); Enumerable<Object[]> scan(DataContext root);  }