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

import java.util.ArrayList;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.LedgerHandle;

/**
 * Driver for a normal Bookkeeper cluster.
 */
class DirectBookkeeperDriver implements BookkeeperVerifier.BookkeeperDriver {
    private final ConcurrentHashMap<Long, LedgerHandle> openHandles = new ConcurrentHashMap<>();
    private BookKeeper client;

    DirectBookkeeperDriver(BookKeeper client) {
        this.client = client;
    }

    @Override
    public void createLedger(long ledgerID, int enSize, int writeQSize, int ackQSize, Consumer<Integer> cb) {
        client.asyncCreateLedgerAdv(
                ledgerID,
                enSize,
                writeQSize,
                ackQSize,
                BookKeeper.DigestType.CRC32,
                new byte[0],
                (rc, lh, ctx) -> {
                    openHandles.put(ledgerID, lh);
                    cb.accept(rc);
                },
                null,
                null);
    }

    @Override
    public void closeLedger(long ledgerID, Consumer<Integer> cb) {
        LedgerHandle handle = openHandles.remove(ledgerID);
        handle.asyncClose(
                (rc, lh, ctx) -> cb.accept(rc),
                null);
    }

    @Override
    public void deleteLedger(long ledgerID, Consumer<Integer> cb) {
        client.asyncDeleteLedger(ledgerID, (rc, ctx) -> {
            cb.accept(rc);
        }, null);
    }

    @Override
    public void writeEntry(long ledgerID, long entryID, byte[] data, Consumer<Integer> cb) {
        LedgerHandle handle;
        handle = openHandles.get(ledgerID);
        if (handle == null) {
            cb.accept(BKException.Code.WriteException);
            return;
        }
        handle.asyncAddEntry(entryID, data, (rc, lh, entryId, ctx) -> {
            cb.accept(rc);
        }, null);
    }

    @Override
    public void readEntries(
            long ledgerID, long firstEntryID, long lastEntryID, BiConsumer<Integer, ArrayList<byte[]>> cb) {
        client.asyncOpenLedgerNoRecovery(ledgerID, BookKeeper.DigestType.CRC32, new byte[0], (rc, lh, ctx) -> {
            if (rc != 0) {
                cb.accept(rc, null);
                return;
            }
            System.out.format("Got handle for read %d -> %d on ledger %d%n", firstEntryID, lastEntryID, ledgerID);
            lh.asyncReadEntries(firstEntryID, lastEntryID, (rc1, lh1, seq, ctx1) -> {
                System.out.format("Read cb %d -> %d on ledger %d%n", firstEntryID, lastEntryID, ledgerID);
                ArrayList<byte[]> results = new ArrayList<>();
                if (rc1 == 0) {
                    while (seq.hasMoreElements()) {
                        results.add(seq.nextElement().getEntry());
                    }
                    System.out.format("About to close handle for read %d -> %d on ledger %d%n",
                            firstEntryID, lastEntryID, ledgerID);
                }
                lh.asyncClose((rc2, lh2, ctx2) -> {
                    System.out.format("Closed handle for read %d -> %d on ledger %d result %d%n",
                            firstEntryID, lastEntryID, ledgerID, rc2);
                    cb.accept(rc1 == 0 ? rc2 : rc1, results);
                }, null);
            }, null);
        }, null);
    }
}
