@Override
    public void onMsg(TbContext ctx, TbMsg msg) {
        EntityType originatorEntityType = msg.getOriginator().getEntityType();
        if (!EntityType.DEVICE.equals(originatorEntityType)) {
            ctx.tellFailure(msg, new IllegalArgumentException(
                    "Unsupported originator entity type: [" + originatorEntityType + "]. Only DEVICE entity type is supported."
            ));
            return;
        }
        DeviceId originator = new DeviceId(msg.getOriginator().getId());
        rateLimits.compute(originator, (__, rateLimit) -> {
            if (rateLimit == null) {
                rateLimit = new TbRateLimits(rateLimitConfig);
            }
            boolean isNotRateLimited = rateLimit.tryConsume();
            if (isNotRateLimited) {
                sendEventAndTell(ctx, originator, msg);
            } else {
                ctx.tellNext(msg, "Rate limited");
            }
            return rateLimit;
        });
    }
@Test
    public void givenMsgArrivedTooFast_whenOnMsg_thenRateLimitsThisMsg() {
        // GIVEN
        ConcurrentReferenceHashMap<DeviceId, TbRateLimits> rateLimits = new ConcurrentReferenceHashMap<>();
        ReflectionTestUtils.setField(node, "rateLimits", rateLimits);

        var rateLimitMock = mock(TbRateLimits.class);
        rateLimits.put(DEVICE_ID, rateLimitMock);

        given(rateLimitMock.tryConsume()).willReturn(false);

        // WHEN
        node.onMsg(ctxMock, msg);

        // THEN
        then(ctxMock).should().tellNext(msg, "Rate limited");
        then(ctxMock).should(never()).tellSuccess(any());
        then(ctxMock).should(never()).tellFailure(any(), any());
        then(ctxMock).shouldHaveNoMoreInteractions();
        then(deviceStateManagerMock).shouldHaveNoInteractions();
    }