bookkeeper-server/src/main/java/org/apache/bookkeeper/client/SyncCallbackUtils.java

/*
 * 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 java.util.Enumeration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.api.LastConfirmedAndEntry;
import org.apache.bookkeeper.client.impl.LastConfirmedAndEntryImpl;

/**
 * Utility for callbacks.
 *
 */
@Slf4j
class SyncCallbackUtils {

    /**
     * Wait for a result. This is convenience method to implement callbacks
     *
     * @param <T>
     * @param future
     * @return
     * @throws InterruptedException
     * @throws BKException
     */
    public static <T> T waitForResult(CompletableFuture<T> future) throws InterruptedException, BKException {
        try {
            try {
                /*
                 * CompletableFuture.get() in JDK8 spins before blocking and wastes CPU time.
                 * CompletableFuture.get(long, TimeUnit) blocks immediately (if the result is
                 * not yet available). While the implementation of get() has changed in JDK9
                 * (not spinning any more), using CompletableFuture.get(long, TimeUnit) allows
                 * us to avoid spinning for all current JDK versions.
                 */
                return future.get(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
            } catch (TimeoutException eignore) {
                // it's ok to return null if we timeout after 292 years (2^63 nanos)
                return null;
            }
        } catch (ExecutionException err) {
            if (err.getCause() instanceof BKException) {
                throw (BKException) err.getCause();
            } else {
                BKException unexpectedConditionException =
                    BKException.create(BKException.Code.UnexpectedConditionException);
                unexpectedConditionException.initCause(err.getCause());
                throw unexpectedConditionException;
            }

        }
    }

    /**
     * Handle the Response Code and transform it to a BKException.
     *
     * @param <T>
     * @param rc
     * @param result
     * @param future
     */
    public static <T> void finish(int rc, T result, CompletableFuture<? super T> future) {
        if (rc != BKException.Code.OK) {
            future.completeExceptionally(BKException.create(rc).fillInStackTrace());
        } else {
            future.complete(result);
        }
    }

    static class SyncCreateCallback implements AsyncCallback.CreateCallback {

        private final CompletableFuture<? super LedgerHandle> future;

        public SyncCreateCallback(CompletableFuture<? super LedgerHandle> future) {
            this.future = future;
        }

        /**
         * Create callback implementation for synchronous create call.
         *
         * @param rc return code
         * @param lh ledger handle object
         * @param ctx optional control object
         */
        @Override
        public void createComplete(int rc, LedgerHandle lh, Object ctx) {
            finish(rc, lh, future);
        }

    }

    static class SyncCreateAdvCallback implements AsyncCallback.CreateCallback {

        private final CompletableFuture<? super LedgerHandleAdv> future;

        public SyncCreateAdvCallback(CompletableFuture<? super LedgerHandleAdv> future) {
            this.future = future;
        }

        /**
         * Create callback implementation for synchronous create call.
         *
         * @param rc return code
         * @param lh ledger handle object
         * @param ctx optional control object
         */
        @Override
        public void createComplete(int rc, LedgerHandle lh, Object ctx) {
            if (lh == null || lh instanceof LedgerHandleAdv) {
                finish(rc, (LedgerHandleAdv) lh, future);
            } else {
                finish(BKException.Code.UnexpectedConditionException, null, future);
            }
        }

    }

    static class SyncOpenCallback implements AsyncCallback.OpenCallback {

        private final CompletableFuture<? super LedgerHandle> future;

        public SyncOpenCallback(CompletableFuture<? super LedgerHandle> future) {
            this.future = future;
        }

        /**
         * Callback method for synchronous open operation.
         *
         * @param rc
         *          return code
         * @param lh
         *          ledger handle
         * @param ctx
         *          optional control object
         */
        @Override
        public void openComplete(int rc, LedgerHandle lh, Object ctx) {
            finish(rc, lh, future);
        }
    }

    static class SyncDeleteCallback implements AsyncCallback.DeleteCallback {

        private final CompletableFuture<Void> future;

        public SyncDeleteCallback(CompletableFuture<Void> future) {
            this.future = future;
        }


        /**
         * Delete callback implementation for synchronous delete call.
         *
         * @param rc
         *            return code
         * @param ctx
         *            optional control object
         */
        @Override
        public void deleteComplete(int rc, Object ctx) {
            finish(rc, null, future);
        }
    }

    static class LastAddConfirmedCallback implements AsyncCallback.AddLacCallback {
        static final LastAddConfirmedCallback INSTANCE = new LastAddConfirmedCallback();
        /**
         * Implementation of callback interface for synchronous read method.
         *
         * @param rc
         *          return code
         * @param lh
         *          ledger identifier
         * @param ctx
         *          control object
         */
        @Override
        public void addLacComplete(int rc, LedgerHandle lh, Object ctx) {
            if (rc != BKException.Code.OK) {
                log.warn("LastAddConfirmedUpdate failed: {} ", BKException.getMessage(rc));
            } else {
                if (log.isDebugEnabled()) {
                    log.debug("Callback LAC Updated for: {} ", lh.getId());
                }
            }
        }
    }

    static class SyncReadCallback implements AsyncCallback.ReadCallback {

        private final CompletableFuture<Enumeration<LedgerEntry>> future;

        public SyncReadCallback(CompletableFuture<Enumeration<LedgerEntry>> future) {
            this.future = future;
        }

        /**
         * Implementation of callback interface for synchronous read method.
         *
         * @param rc
         *          return code
         * @param lh
         *          ledger handle
         * @param seq
         *          sequence of entries
         * @param ctx
         *          control object
         */
        @Override
        public void readComplete(int rc, LedgerHandle lh,
                                 Enumeration<LedgerEntry> seq, Object ctx) {
            finish(rc, seq, future);
        }
    }

    static class SyncAddCallback extends CompletableFuture<Long> implements AsyncCallback.AddCallback {

        /**
         * Implementation of callback interface for synchronous read method.
         *
         * @param rc
         *          return code
         * @param lh
         *          ledger handle
         * @param entry
         *          entry identifier
         * @param ctx
         *          control object
         */
        @Override
        public void addComplete(int rc, LedgerHandle lh, long entry, Object ctx) {
            finish(rc, entry, this);
        }
    }

    static class FutureReadLastConfirmed extends CompletableFuture<Long>
        implements AsyncCallback.ReadLastConfirmedCallback {

        @Override
        public void readLastConfirmedComplete(int rc, long lastConfirmed, Object ctx) {
            finish(rc, lastConfirmed, this);
        }

    }

    static class SyncReadLastConfirmedCallback implements AsyncCallback.ReadLastConfirmedCallback {
        /**
         * Implementation of  callback interface for synchronous read last confirmed method.
         */
        @Override
        public void readLastConfirmedComplete(int rc, long lastConfirmed, Object ctx) {
            LedgerHandle.LastConfirmedCtx lcCtx = (LedgerHandle.LastConfirmedCtx) ctx;

            synchronized (lcCtx) {
                lcCtx.setRC(rc);
                lcCtx.setLastConfirmed(lastConfirmed);
                lcCtx.notify();
            }
        }
    }

    static class SyncCloseCallback implements AsyncCallback.CloseCallback {

        private final CompletableFuture<Void> future;

        public SyncCloseCallback(CompletableFuture<Void> future) {
            this.future = future;
        }

        /**
         * Close callback method.
         *
         * @param rc
         * @param lh
         * @param ctx
         */
        @Override
        public void closeComplete(int rc, LedgerHandle lh, Object ctx) {
            finish(rc, null, future);
        }
    }

    static class FutureReadLastConfirmedAndEntry
        extends CompletableFuture<LastConfirmedAndEntry> implements AsyncCallback.ReadLastConfirmedAndEntryCallback {

        @Override
        public void readLastConfirmedAndEntryComplete(int rc, long lastConfirmed, LedgerEntry entry, Object ctx) {
            LastConfirmedAndEntry result = LastConfirmedAndEntryImpl.create(lastConfirmed, entry);
            finish(rc, result, this);
        }
    }

}