WindowUtils { public static long tpsToTpw(OperatorContext context, long tuplesPerSecond) { Preconditions.checkArgument(tuplesPerSecond > 0); BigDecimal tuplesPerWindow = new BigDecimal(getAppWindowDurationMs(context)); tuplesPerWindow = tuplesPerWindow.divide(new BigDecimal(1000)); tuplesPerWindow = tuplesPerWindow.multiply(new BigDecimal(tuplesPerSecond)); Preconditions.checkArgument(tuplesPerWindow.compareTo(new BigDecimal(Long.MAX_VALUE)) <= 0, "Overflow computing tuples per window."); tuplesPerWindow = tuplesPerWindow.stripTrailingZeros(); long tuplesPerWindowLong = tuplesPerWindow.longValue(); if (tuplesPerWindow.scale() > 0) { LOG.debug("{}", tuplesPerWindow); tuplesPerWindowLong++; } return tuplesPerWindowLong; }  static long getAppWindowDurationMs(OperatorContext context); static long msToAppWindowCount(OperatorContext context, long millis); static long tpsToTpw(OperatorContext context, long tuplesPerSecond);  }
@Test public void tpsToTpwSimpleTest() { OperatorContext context = createOperatorContext(100, 1); long tuplesPerWindow = WindowUtils.tpsToTpw(context, 500L); Assert.assertEquals(50L, tuplesPerWindow); }
@Test public void tpsToTpwRoundingTest() { OperatorContext context = createOperatorContext(100, 1); long tuplesPerWindow = WindowUtils.tpsToTpw(context, 501L); Assert.assertEquals(51L, tuplesPerWindow); }
@Test public void tpsToTpwOverflowTest() { OperatorContext context = createOperatorContext(Integer.MAX_VALUE, 1); boolean overflowException = false; try { WindowUtils.tpsToTpw(context, Long.MAX_VALUE); } catch (Exception e) { overflowException = true; } Assert.assertTrue(overflowException); }