@Test public void testUnionOperator() throws InjectionException { final MistDataEvent a = new MistDataEvent("a", 1L); final MistDataEvent b = new MistDataEvent("b", 3L); final MistDataEvent c = new MistDataEvent("c", 5L); final MistWatermarkEvent lw1 = new MistWatermarkEvent(9L); final MistWatermarkEvent lw2 = new MistWatermarkEvent(10L); final MistDataEvent d = new MistDataEvent("d", 1L); final MistDataEvent e = new MistDataEvent("e", 2L); final MistDataEvent f = new MistDataEvent("f", 4L); final MistWatermarkEvent rw1 = new MistWatermarkEvent(3L); final MistDataEvent g = new MistDataEvent("g", 6L); final MistWatermarkEvent rw2 = new MistWatermarkEvent(11L); final UnionOperator unionOperator = new UnionOperator(); final List<MistEvent> result = new LinkedList<>(); unionOperator.setOutputEmitter(new OutputBufferEmitter(result)); unionOperator.processLeftData(a); Assert.assertEquals(0, result.size()); unionOperator.processRightData(d); Assert.assertEquals(2, result.size()); Assert.assertEquals(a, result.get(0)); Assert.assertEquals(d, result.get(1)); unionOperator.processRightData(e); unionOperator.processRightData(f); Assert.assertEquals(2, result.size()); unionOperator.processLeftData(b); Assert.assertEquals(4, result.size()); Assert.assertEquals(e, result.get(2)); Assert.assertEquals(b, result.get(3)); unionOperator.processRightWatermark(rw1); Assert.assertEquals(5, result.size()); Assert.assertEquals(rw1, result.get(4)); unionOperator.processLeftData(c); Assert.assertEquals(6, result.size()); Assert.assertEquals(f, result.get(5)); unionOperator.processLeftWatermark(lw1); Assert.assertEquals(6, result.size()); unionOperator.processLeftWatermark(lw2); Assert.assertEquals(6, result.size()); unionOperator.processRightData(g); Assert.assertEquals(8, result.size()); Assert.assertEquals(c, result.get(6)); Assert.assertEquals(g, result.get(7)); unionOperator.processRightWatermark(rw2); Assert.assertEquals(9, result.size()); Assert.assertEquals(lw2, result.get(8)); }
@Inject public UnionOperator() { this.leftUpstreamQueue = new LinkedBlockingQueue<>(); this.rightUpstreamQueue = new LinkedBlockingQueue<>(); defaultWatermark = new MistWatermarkEvent(0L); this.latestLeftWatermark = defaultWatermark; this.latestRightWatermark = defaultWatermark; this.recentLeftTimestamp = 0L; this.recentRightTimestamp = 0L; }
UnionOperator extends TwoStreamOperator { @Inject public UnionOperator() { this.leftUpstreamQueue = new LinkedBlockingQueue<>(); this.rightUpstreamQueue = new LinkedBlockingQueue<>(); defaultWatermark = new MistWatermarkEvent(0L); this.latestLeftWatermark = defaultWatermark; this.latestRightWatermark = defaultWatermark; this.recentLeftTimestamp = 0L; this.recentRightTimestamp = 0L; } }
UnionOperator extends TwoStreamOperator { @Inject public UnionOperator() { this.leftUpstreamQueue = new LinkedBlockingQueue<>(); this.rightUpstreamQueue = new LinkedBlockingQueue<>(); defaultWatermark = new MistWatermarkEvent(0L); this.latestLeftWatermark = defaultWatermark; this.latestRightWatermark = defaultWatermark; this.recentLeftTimestamp = 0L; this.recentRightTimestamp = 0L; } @Inject  UnionOperator(); }
UnionOperator extends TwoStreamOperator { @Inject public UnionOperator() { this.leftUpstreamQueue = new LinkedBlockingQueue<>(); this.rightUpstreamQueue = new LinkedBlockingQueue<>(); defaultWatermark = new MistWatermarkEvent(0L); this.latestLeftWatermark = defaultWatermark; this.latestRightWatermark = defaultWatermark; this.recentLeftTimestamp = 0L; this.recentRightTimestamp = 0L; } @Inject  UnionOperator(); @Override void processLeftData(final MistDataEvent event); @Override void processRightData(final MistDataEvent event); @Override void processLeftWatermark(final MistWatermarkEvent event); @Override void processRightWatermark(final MistWatermarkEvent event); @Override void processLeftCheckpoint(final MistCheckpointEvent input); @Override void processRightCheckpoint(final MistCheckpointEvent input); }
UnionOperator extends TwoStreamOperator { @Inject public UnionOperator() { this.leftUpstreamQueue = new LinkedBlockingQueue<>(); this.rightUpstreamQueue = new LinkedBlockingQueue<>(); defaultWatermark = new MistWatermarkEvent(0L); this.latestLeftWatermark = defaultWatermark; this.latestRightWatermark = defaultWatermark; this.recentLeftTimestamp = 0L; this.recentRightTimestamp = 0L; } @Inject  UnionOperator(); @Override void processLeftData(final MistDataEvent event); @Override void processRightData(final MistDataEvent event); @Override void processLeftWatermark(final MistWatermarkEvent event); @Override void processRightWatermark(final MistWatermarkEvent event); @Override void processLeftCheckpoint(final MistCheckpointEvent input); @Override void processRightCheckpoint(final MistCheckpointEvent input);  }