class StormTopologyTimeSeriesMetricsImpl implements TopologyTimeSeriesMetrics {

    @Override
    public TimeSeriesComponentMetric getComponentStats(TopologyLayout topology, Component component, long from, long to, String asUser) {
        String stormTopologyName = StormTopologyUtil.findOrGenerateTopologyName(client, topology.getId(), topology.getName(), asUser);
        String stormComponentName = getComponentName(component);

        StormMappedMetric[] stats;
        if (component instanceof Source) {
            stats = STATS_METRICS_SOURCE;
        } else {
            stats = STATS_METRICS;
        }

        // empty map if time-series DB is not set to the namespace
        Map<String, Map<Long, Double>> componentStats = ParallelStreamUtil.execute(() ->
                        Arrays.asList(stats)
                                .parallelStream()
                                .collect(toMap(m -> m.name(),
                                        m -> queryComponentMetrics(stormTopologyName, stormComponentName, m, from, to))),
                FORK_JOIN_POOL);

        return buildTimeSeriesComponentMetric(component.getName(), componentStats);
    }

    public  StormTopologyTimeSeriesMetricsImpl(StormRestAPIClient client);

}