class DeviceMessageMeasurementProvider extends StaticMeasurementProvider {

    MeterRegistry registry;

    public DeviceMessageMeasurementProvider(EventBus eventBus,
                                            MeterRegistryManager registryManager,
                                            TimeSeriesManager timeSeriesManager) {
        super(DeviceDashboardDefinition.instance, DeviceObjectDefinition.message);

        registry = registryManager.getMeterRegister(DeviceTimeSeriesMetric.deviceMetrics().getId(),
            "target", "msgType", "productId");

        addMeasurement(new DeviceMessageMeasurement(eventBus, timeSeriesManager));

    }

    @Subscribe("/device/*/*/message/**")
    public Mono<Void> incrementMessage(DeviceMessage message) {
        return Mono.fromRunnable(() -> {
            registry
                .counter("message-count", convertTags(message))
                .increment();
        });
    }

    static final String[] empty = new String[0];

    private String[] convertTags(DeviceMessage message) {<FILL_FUNCTION_BODY>}
}

        if (message == null) {
            return empty;
        }
        return new String[]{
            "productId", message.getHeader("productId").map(String::valueOf).orElse("unknown")
        };
    
<methods>public void <init>(org.jetlinks.community.dashboard.DashboardDefinition, org.jetlinks.community.dashboard.ObjectDefinition) ,public Mono<org.jetlinks.community.dashboard.Measurement> getMeasurement(java.lang.String) ,public Flux<org.jetlinks.community.dashboard.Measurement> getMeasurements() <variables>private org.jetlinks.community.dashboard.DashboardDefinition dashboardDefinition,private Map<java.lang.String,org.jetlinks.community.dashboard.Measurement> measurements,private org.jetlinks.community.dashboard.ObjectDefinition objectDefinition