class ElementCountFnDataReceiver implements FnDataReceiver<WindowedValue<T>> {

  @Override
  public void accept(WindowedValue<T> input) throws Exception {
    try (Closeable close = MetricsEnvironment.scopedMetricsContainer(this.unboundMetricContainer)) {
      // Increment the counter for each window the element occurs in.
      this.counter.inc(input.getWindows().size());
      this.original.accept(input);
    }
  }

  public  ElementCountFnDataReceiver(
      FnDataReceiver<WindowedValue<T>> original,
      String pCollection,
      MetricsContainerStepMap metricContainerRegistry);

}