StatusWatermarkValve { public void inputStreamStatus(StreamStatus streamStatus, int channelIndex) { if (streamStatus.isIdle() && channelStatuses[channelIndex].streamStatus.isActive()) { channelStatuses[channelIndex].streamStatus = StreamStatus.IDLE; channelStatuses[channelIndex].isWatermarkAligned = false; if (!InputChannelStatus.hasActiveChannels(channelStatuses)) { if (channelStatuses[channelIndex].watermark == lastOutputWatermark) { findAndOutputMaxWatermarkAcrossAllChannels(); } lastOutputStreamStatus = StreamStatus.IDLE; outputHandler.handleStreamStatus(lastOutputStreamStatus); } else if (channelStatuses[channelIndex].watermark == lastOutputWatermark) { findAndOutputNewMinWatermarkAcrossAlignedChannels(); } } else if (streamStatus.isActive() && channelStatuses[channelIndex].streamStatus.isIdle()) { channelStatuses[channelIndex].streamStatus = StreamStatus.ACTIVE; if (channelStatuses[channelIndex].watermark >= lastOutputWatermark) { channelStatuses[channelIndex].isWatermarkAligned = true; } if (lastOutputStreamStatus.isIdle()) { lastOutputStreamStatus = StreamStatus.ACTIVE; outputHandler.handleStreamStatus(lastOutputStreamStatus); } } } StatusWatermarkValve(int numInputChannels, ValveOutputHandler outputHandler); void inputWatermark(Watermark watermark, int channelIndex); void inputStreamStatus(StreamStatus streamStatus, int channelIndex);  }
@Test public void testSingleInputStreamStatusToggling() { BufferedValveOutputHandler valveOutput = new BufferedValveOutputHandler(); StatusWatermarkValve valve = new StatusWatermarkValve(1, valveOutput); valve.inputStreamStatus(StreamStatus.ACTIVE, 0); assertEquals(null, valveOutput.popLastSeenOutput()); valve.inputStreamStatus(StreamStatus.IDLE, 0); assertEquals(StreamStatus.IDLE, valveOutput.popLastSeenOutput()); valve.inputStreamStatus(StreamStatus.IDLE, 0); assertEquals(null, valveOutput.popLastSeenOutput()); valve.inputStreamStatus(StreamStatus.ACTIVE, 0); assertEquals(StreamStatus.ACTIVE, valveOutput.popLastSeenOutput()); assertEquals(null, valveOutput.popLastSeenOutput()); }
@Test public void testMultipleInputStreamStatusToggling() { BufferedValveOutputHandler valveOutput = new BufferedValveOutputHandler(); StatusWatermarkValve valve = new StatusWatermarkValve(2, valveOutput); valve.inputStreamStatus(StreamStatus.ACTIVE, 0); valve.inputStreamStatus(StreamStatus.ACTIVE, 1); assertEquals(null, valveOutput.popLastSeenOutput()); valve.inputStreamStatus(StreamStatus.IDLE, 1); assertEquals(null, valveOutput.popLastSeenOutput()); valve.inputStreamStatus(StreamStatus.IDLE, 0); assertEquals(StreamStatus.IDLE, valveOutput.popLastSeenOutput()); valve.inputStreamStatus(StreamStatus.IDLE, 0); valve.inputStreamStatus(StreamStatus.IDLE, 1); assertEquals(null, valveOutput.popLastSeenOutput()); valve.inputStreamStatus(StreamStatus.ACTIVE, 1); assertEquals(StreamStatus.ACTIVE, valveOutput.popLastSeenOutput()); valve.inputStreamStatus(StreamStatus.ACTIVE, 0); assertEquals(null, valveOutput.popLastSeenOutput()); }