/*
 * Copyright 2016 higherfrequencytrading.com
 *
 * 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 net.openhft.chronicle.queue.impl;

import net.openhft.chronicle.core.annotation.Nullable;
import net.openhft.chronicle.queue.RollDetails;
import net.openhft.chronicle.queue.TailerDirection;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.ref.Reference;
import java.lang.ref.WeakReference;
import java.text.ParseException;
import java.util.Map;
import java.util.NavigableSet;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

public class WireStorePool {
    private static final Logger LOGGER = LoggerFactory.getLogger(WireStorePool.class);
    // must be power-of-two
    private static final int ROLL_CYCLE_CACHE_SIZE = 64;
    private static final int INDEX_MASK = ROLL_CYCLE_CACHE_SIZE - 1;
    @NotNull
    private final WireStoreSupplier supplier;
    @NotNull
    private final Map<RollDetails, WeakReference<WireStore>> stores;
    private final StoreFileListener storeFileListener;
    private boolean isClosed = false;
    // protected by synchronized on acquire()
    private final RollDetails[] cache = new RollDetails[ROLL_CYCLE_CACHE_SIZE];


    private WireStorePool(@NotNull WireStoreSupplier supplier, StoreFileListener storeFileListener) {
        this.supplier = supplier;
        this.storeFileListener = storeFileListener;
        this.stores = new ConcurrentHashMap<>();
    }

    @NotNull
    public static WireStorePool withSupplier(@NotNull WireStoreSupplier supplier, StoreFileListener storeFileListener) {
        return new WireStorePool(supplier, storeFileListener);
    }

    public synchronized void close() {
        if (isClosed)
            return;
        isClosed = true;

        stores.values().stream()
                .map(Reference::get)
                .filter(Objects::nonNull)
                .forEach(this::release);
    }

    @org.jetbrains.annotations.Nullable
    @Nullable
    public synchronized WireStore acquire(final int cycle, final long epoch, boolean createIfAbsent) {
        final int cacheIndex = cacheIndex(cycle);
        RollDetails rollDetails;
        rollDetails = cache[cacheIndex];
        if (rollDetails == null || rollDetails.cycle() != cycle) {
            rollDetails = new RollDetails(cycle, epoch);
            cache[cacheIndex] = rollDetails;
        }

        WeakReference<WireStore> reference = stores.get(rollDetails);
        WireStore store;
        if (reference != null) {
            store = reference.get();
            if (store != null) {
                if (store.tryReserve())
                    return store;
                else
                    /// this should never happen,
                    // this method is synchronized
                    // and this remove below, is only any use if the acquire method below that fails
                    stores.remove(rollDetails);
            }
        }

        store = this.supplier.acquire(cycle, createIfAbsent);
        if (store != null) {
            stores.put(rollDetails, new WeakReference<>(store));
            storeFileListener.onAcquired(cycle, store.file());
        }
        return store;
    }

    public int nextCycle(final int currentCycle, @NotNull TailerDirection direction) throws ParseException {
        return supplier.nextCycle(currentCycle, direction);
    }

    public synchronized void release(@NotNull CommonStore store) {
        store.release();

        long refCount = store.refCount();
        assert refCount >= 0;
        if (refCount == 0) {
            for (Map.Entry<RollDetails, WeakReference<WireStore>> entry : stores.entrySet()) {
                WeakReference<WireStore> ref = entry.getValue();
                if (ref != null && ref.get() == store) {
                    stores.remove(entry.getKey());
                    storeFileListener.onReleased(entry.getKey().cycle(), store.file());
                    return;
                }
            }
            LOGGER.warn("Store was not registered {}", store.file().getName());
        }
    }

    /**
     * list cycles between ( inclusive )
     *
     * @param lowerCycle the lower cycle
     * @param upperCycle the upper cycle
     * @return an array including these cycles and all the intermediate cycles
     */
    public NavigableSet<Long> listCyclesBetween(int lowerCycle, int upperCycle) throws ParseException {
        return supplier.cycles(lowerCycle, upperCycle);
    }

    public boolean isEmpty() {
        return stores.isEmpty();
    }

    private static int cacheIndex(final int cycle) {
        return cycle & INDEX_MASK;
    }
}

@@ -646,16 +646,55 @@
 .impl;%0A%0A
+import net.openhft.chronicle.core.Jvm;%0A
 import n
@@ -740,16 +740,16 @@
 llable;%0A
-
 import n
@@ -885,65 +885,8 @@
 ull;
-%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;
 %0A%0Aim
@@ -1144,95 +1144,8 @@
 l %7B%0A
-    private static final Logger LOGGER = LoggerFactory.getLogger(WireStorePool.class);%0A
     
@@ -1574,16 +1574,16 @@
 quire()%0A
+
     priv
@@ -1655,17 +1655,16 @@
 SIZE%5D;%0A%0A
-%0A
     priv
@@ -4459,20 +4459,90 @@
     
-LOGGER.warn(
+if (Jvm.debug().isEnabled(getClass()))%0A                Jvm.debug().on(getClass(), 
 %22Sto
@@ -4566,13 +4566,13 @@
 ered
- %7B%7D%22,
+: %22 +
  sto
@@ -4584,18 +4584,8 @@
 le()
-.getName()
 );%0A 

Downgrade log statement #449