/*
 *
 * 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.bookkeeper.client;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.bookkeeper.client.AsyncCallback.AddCallback;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Test a quorum timeout for add entry operations.
 */
public class TestAddEntryQuorumTimeout extends BookKeeperClusterTestCase implements AddCallback {

    private static final Logger logger = LoggerFactory.getLogger(TestAddEntryQuorumTimeout.class);

    final DigestType digestType;
    final byte[] testPasswd = "".getBytes();

    public TestAddEntryQuorumTimeout() {
        super(3);
        baseClientConf.setAddEntryTimeout(10);
        baseClientConf.setAddEntryQuorumTimeout(1);
        this.digestType = DigestType.CRC32;
    }

    @Before
    @Override
    public void setUp() throws Exception {
        super.setUp();
        baseConf.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
    }

    private static class SyncObj {
        volatile int counter = 0;
        volatile int rc = -1;
        public SyncObj() {
            counter = 0;
        }
    }

    @Override
    public void addComplete(int rc, LedgerHandle lh, long entryId, Object ctx) {
        SyncObj x = (SyncObj) ctx;
        synchronized (x) {
            x.rc = rc;
            x.counter++;
            x.notify();
        }
    }

    @Test
    public void testBasicTimeout() throws Exception {
        BookKeeperTestClient bkc = new BookKeeperTestClient(baseClientConf);
        LedgerHandle lh = bkc.createLedger(3, 3, 3, digestType, testPasswd);
        List<BookieSocketAddress> curEns = lh.getCurrentEnsemble();
        byte[] data = "foobar".getBytes();
        lh.addEntry(data);
        sleepBookie(curEns.get(0), 5).await();
        try {
            lh.addEntry(data);
            fail("should have thrown");
        } catch (BKException.BKAddEntryQuorumTimeoutException ex) {
        }
    }

    private void waitForSyncObj(SyncObj syncObj) throws Exception {
        synchronized (syncObj) {
            while (syncObj.counter < 1) {
                logger.debug("Entries counter = " + syncObj.counter);
                syncObj.wait();
            }
        }
    }

    @Test
    public void testTimeoutWithPendingOps() throws Exception {
        BookKeeperTestClient bkc = new BookKeeperTestClient(baseClientConf);
        LedgerHandle lh = bkc.createLedger(3, 3, 3, digestType, testPasswd);
        List<BookieSocketAddress> curEns = lh.getCurrentEnsemble();
        byte[] data = "foobar".getBytes();

        SyncObj syncObj1 = new SyncObj();
        SyncObj syncObj2 = new SyncObj();
        SyncObj syncObj3 = new SyncObj();

        lh.addEntry(data);
        sleepBookie(curEns.get(0), 5).await();
        lh.asyncAddEntry(data, this, syncObj1);
        lh.asyncAddEntry(data, this, syncObj2);
        lh.asyncAddEntry(data, this, syncObj3);

        waitForSyncObj(syncObj1);
        assertEquals(BKException.Code.AddEntryQuorumTimeoutException, syncObj1.rc);
        waitForSyncObj(syncObj2);
        assertEquals(BKException.Code.AddEntryQuorumTimeoutException, syncObj2.rc);
        waitForSyncObj(syncObj3);
        assertEquals(BKException.Code.AddEntryQuorumTimeoutException, syncObj3.rc);
    }

    @Test
    public void testLedgerClosedAfterTimeout() throws Exception {
        BookKeeperTestClient bkc = new BookKeeperTestClient(baseClientConf);
        LedgerHandle lh = bkc.createLedger(3, 3, 3, digestType, testPasswd);
        List<BookieSocketAddress> curEns = lh.getCurrentEnsemble();
        byte[] data = "foobar".getBytes();
        CountDownLatch b0latch = sleepBookie(curEns.get(0), 5);
        try {
            lh.addEntry(data);
            fail("should have thrown");
        } catch (BKException.BKAddEntryQuorumTimeoutException ex) {
        }
        b0latch.await();
        try {
            lh.addEntry(data);
            fail("should have thrown");
        } catch (BKException.BKLedgerClosedException ex) {
        }
    }
}