/*
 * Copyright 2011-2017 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.lettuce.core;

import static io.lettuce.core.AbstractRedisClientTest.client;
import static org.assertj.core.api.Assertions.assertThat;

import java.util.Collection;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.test.util.ReflectionTestUtils;

import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import io.lettuce.TestClientResources;
import io.lettuce.Wait;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.event.EventBus;
import io.lettuce.core.event.metrics.CommandLatencyEvent;
import io.lettuce.core.event.metrics.MetricEventPublisher;
import io.lettuce.core.metrics.CommandLatencyId;
import io.lettuce.core.metrics.CommandMetrics;

/**
 * @author Mark Paluch
 */
public class ClientMetricsTest extends AbstractTest {

    private RedisCommands<String, String> redis;

    @BeforeClass
    public static void setupClient() {
        client = RedisClient.create(TestClientResources.get(), RedisURI.Builder.redis(host, port).build());
    }

    @Before
    public void before() throws Exception {
        redis = client.connect().sync();
    }

    @AfterClass
    public static void afterClass() {
        FastShutdown.shutdown(client);
    }

    @Test
    public void testMetricsEvent() throws Exception {

        Collection<CommandLatencyEvent> events = new LinkedBlockingQueue<>();
        EventBus eventBus = client.getResources().eventBus();
        MetricEventPublisher publisher = (MetricEventPublisher) ReflectionTestUtils.getField(client.getResources(),
                "metricEventPublisher");
        publisher.emitMetricsEvent();

        Disposable disposable = eventBus.get().filter(redisEvent -> redisEvent instanceof CommandLatencyEvent)
                .cast(CommandLatencyEvent.class).doOnNext(events::add).subscribe();

        generateTestData();
        publisher.emitMetricsEvent();

        Wait.untilTrue(() -> !events.isEmpty()).waitOrTimeout();

        assertThat(events).isNotEmpty();

        disposable.dispose();
    }

    @Test
    public void testMetrics() throws Exception {

        EventBus eventBus = client.getResources().eventBus();
        MetricEventPublisher publisher = (MetricEventPublisher) ReflectionTestUtils.getField(client.getResources(),
                "metricEventPublisher");

        Flux<CommandLatencyEvent> flux = eventBus.get().filter(redisEvent -> redisEvent instanceof CommandLatencyEvent)
                .cast(CommandLatencyEvent.class);

        StepVerifier.create(flux).then(() -> {
            generateTestData();
            publisher.emitMetricsEvent();
        }).consumeNextWith(actual -> {
            // just consume
            }).consumeNextWith(actual -> {

            Set<CommandLatencyId> ids = actual.getLatencies().keySet();
            CommandMetrics commandMetrics = actual.getLatencies().get(ids.iterator().next());
            assertThat(commandMetrics.getCompletion().getMin()).isBetween(0L, TimeUnit.MILLISECONDS.toMicros(100));
            assertThat(commandMetrics.getCompletion().getMax()).isBetween(0L, TimeUnit.MILLISECONDS.toMicros(200));

            assertThat(commandMetrics.getFirstResponse().getMin()).isBetween(0L, TimeUnit.MILLISECONDS.toMicros(100));
            assertThat(commandMetrics.getFirstResponse().getMax()).isBetween(0L, TimeUnit.MILLISECONDS.toMicros(200));
        }).thenCancel().verify();

    }

    private void generateTestData() {
        redis.set(key, value);
        redis.set(key, value);
        redis.set(key, value);
        redis.set(key, value);
        redis.set(key, value);
        redis.set(key, value);

        redis.get(key);
        redis.get(key);
        redis.get(key);
        redis.get(key);
        redis.get(key);
    }
}

@@ -794,30 +794,8 @@
 on;%0A
-import java.util.Set;%0A
 impo
@@ -842,46 +842,8 @@
 eue;
-%0Aimport java.util.concurrent.TimeUnit;
 %0A%0Aim
@@ -1042,78 +1042,8 @@
 le;%0A
-import reactor.core.publisher.Flux;%0Aimport reactor.test.StepVerifier;%0A
 impo
@@ -1307,104 +1307,8 @@
 her;
-%0Aimport io.lettuce.core.metrics.CommandLatencyId;%0Aimport io.lettuce.core.metrics.CommandMetrics;
 %0A%0A/*
@@ -2584,24 +2584,24 @@
 otEmpty();%0A%0A
+
         disp
@@ -2629,1370 +2629,8 @@
  %7D%0A%0A
-    @Test%0A    public void testMetrics() throws Exception %7B%0A%0A        EventBus eventBus = client.getResources().eventBus();%0A        MetricEventPublisher publisher = (MetricEventPublisher) ReflectionTestUtils.getField(client.getResources(),%0A                %22metricEventPublisher%22);%0A%0A        Flux%3CCommandLatencyEvent%3E flux = eventBus.get().filter(redisEvent -%3E redisEvent instanceof CommandLatencyEvent)%0A                .cast(CommandLatencyEvent.class);%0A%0A        StepVerifier.create(flux).then(() -%3E %7B%0A            generateTestData();%0A            publisher.emitMetricsEvent();%0A        %7D).consumeNextWith(actual -%3E %7B%0A            // just consume%0A            %7D).consumeNextWith(actual -%3E %7B%0A%0A            Set%3CCommandLatencyId%3E ids = actual.getLatencies().keySet();%0A            CommandMetrics commandMetrics = actual.getLatencies().get(ids.iterator().next());%0A            assertThat(commandMetrics.getCompletion().getMin()).isBetween(0L, TimeUnit.MILLISECONDS.toMicros(100));%0A            assertThat(commandMetrics.getCompletion().getMax()).isBetween(0L, TimeUnit.MILLISECONDS.toMicros(200));%0A%0A            assertThat(commandMetrics.getFirstResponse().getMin()).isBetween(0L, TimeUnit.MILLISECONDS.toMicros(100));%0A            assertThat(commandMetrics.getFirstResponse().getMax()).isBetween(0L, TimeUnit.MILLISECONDS.toMicros(200));%0A        %7D).thenCancel().verify();%0A%0A    %7D%0A%0A
     

Disable metrics tests