/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.jackrabbit.oak.plugins.observation;

import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.apache.jackrabbit.oak.api.CommitFailedException;
import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
import org.apache.jackrabbit.oak.spi.state.NodeState;
import org.apache.jackrabbit.oak.stats.Clock;
import org.junit.Before;
import org.junit.Test;

public class CommitRateLimiterTest {
    private static final NodeState AFTER = createAfter();

    private static NodeState createAfter() {
        NodeBuilder builder = EMPTY_NODE.builder();
        builder.setChildNode("a");
        builder.setProperty("x", 42);
        return builder.getNodeState();
    }

    private CommitRateLimiter limiter;

    @Before
    public void setup() {
        limiter = new CommitRateLimiter();
    }

    @Test
    public void commit() throws CommitFailedException {
        assertSame(AFTER, limiter.processCommit(EMPTY_NODE, AFTER, null));
    }

    @Test(expected = CommitFailedException.class)
    public void blockCommits() throws CommitFailedException {
        limiter.blockCommits();
        limiter.processCommit(EMPTY_NODE, AFTER, null);
    }

    @Test
    public void delayCommits() throws CommitFailedException {
        limiter.setDelay(1000);
        long t0 = Clock.ACCURATE.getTime();
        assertSame(AFTER, limiter.processCommit(EMPTY_NODE, AFTER, null));
        assertTrue(Clock.ACCURATE.getTime() - t0 >= 1000);
    }

    private final FutureTask<Long> commit = new FutureTask<Long>(new Callable<Long>() {
        @Override
        public Long call() throws Exception {
            long t0 = Clock.ACCURATE.getTime();
            Clock.ACCURATE.waitUntil(Clock.ACCURATE.getTime() + 100);
            assertSame(AFTER, limiter.processCommit(EMPTY_NODE, AFTER, null));
            return Clock.ACCURATE.getTime() - t0;
        }
    });

    @Test
    public void delayCommitsWithReset() throws InterruptedException, ExecutionException, TimeoutException {
        limiter.setDelay(10000);
        new Thread(commit).start();
        limiter.setDelay(0);
        assertTrue(commit.get(1, TimeUnit.SECONDS) >= 100);
    }

    @Test(expected = ExecutionException.class)
    public void delayCommitsWithInterrupt() throws InterruptedException, ExecutionException, TimeoutException {
        limiter.setDelay(10000);
        Thread t = new Thread(commit);
        t.start();
        t.interrupt();
        commit.get(1, TimeUnit.SECONDS);
    }
}
