/*
 * Copyright 2014 - 2016 Real Logic Ltd.
 *
 * 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 uk.co.real_logic.agrona.concurrent;

import uk.co.real_logic.agrona.BitUtil;

import java.util.Deque;
import java.util.LinkedList;
import java.util.function.BiConsumer;

import static uk.co.real_logic.agrona.BitUtil.SIZE_OF_INT;

/**
 * Manages the allocation and freeing of counters.
 */
public class CountersManager
{
    public static final int LABEL_LENGTH = BitUtil.CACHE_LINE_LENGTH * 2;
    public static final int COUNTER_LENGTH = BitUtil.CACHE_LINE_LENGTH * 2;
    public static final int UNREGISTERED_LABEL_LENGTH = -1;

    private final AtomicBuffer labelsBuffer;
    private final AtomicBuffer countersBuffer;
    private final Deque<Integer> freeList = new LinkedList<>();

    private int idHighWaterMark = -1;

    /**
     * Create a new counter buffer manager over two buffers.
     *
     * @param labelsBuffer containing the human readable labels for the counters.
     * @param countersBuffer containing the values of the counters themselves.
     */
    public CountersManager(final AtomicBuffer labelsBuffer, final AtomicBuffer countersBuffer)
    {
        this.labelsBuffer = labelsBuffer;
        this.countersBuffer = countersBuffer;
        countersBuffer.verifyAlignment();
    }

    /**
     * Allocate a new counter with a given label.
     *
     * @param label to describe the counter.
     * @return the id allocated for the counter.
     */
    public int allocate(final String label)
    {
        final int counterId = counterId();
        final int labelsOffset = labelOffset(counterId);
        if ((counterOffset(counterId) + COUNTER_LENGTH) > countersBuffer.capacity())
        {
            throw new IllegalArgumentException("Unable to allocated counter, counter buffer is full");
        }

        if ((labelsOffset + LABEL_LENGTH) > labelsBuffer.capacity())
        {
            throw new IllegalArgumentException("Unable to allocate counter, labels buffer is full");
        }

        labelsBuffer.putStringUtf8(labelsOffset, label, LABEL_LENGTH - SIZE_OF_INT);

        return counterId;
    }

    public AtomicCounter newCounter(final String label)
    {
        return new AtomicCounter(countersBuffer, allocate(label), this);
    }

    /**
     * Free the counter identified by counterId.
     *
     * @param counterId the counter to freed
     */
    public void free(final int counterId)
    {
        labelsBuffer.putInt(labelOffset(counterId), UNREGISTERED_LABEL_LENGTH);
        freeList.push(counterId);
    }

    /**
     * The offset in the counter buffer for a given id.
     *
     * @param id for which the offset should be provided.
     * @return the offset in the counter buffer.
     */
    public static int counterOffset(int id)
    {
        return id * COUNTER_LENGTH;
    }

    /**
     * Iterate over all labels in the label buffer.
     *
     * @param consumer function to be called for each label.
     */
    public void forEach(final BiConsumer<Integer, String> consumer)
    {
        int labelsOffset = 0;
        int size;
        int id = 0;

        while ((size = labelsBuffer.getInt(labelsOffset)) != 0)
        {
            if (size != UNREGISTERED_LABEL_LENGTH)
            {
                final String label = labelsBuffer.getStringUtf8(labelsOffset);
                consumer.accept(id, label);
            }

            labelsOffset += LABEL_LENGTH;
            id++;
        }
    }

    /**
     * Set an {@link AtomicCounter} value based on counterId.
     *
     * @param counterId to be set.
     * @param value to set for the counter.
     */
    public void setCounterValue(final int counterId, final long value)
    {
        countersBuffer.putLongOrdered(counterOffset(counterId), value);
    }

    private static int labelOffset(final int counterId)
    {
        return counterId * LABEL_LENGTH;
    }

    private int counterId()
    {
        if (freeList.isEmpty())
        {
            return ++idHighWaterMark;
        }

        final int counterId = freeList.pop();
        countersBuffer.putLongOrdered(counterOffset(counterId), 0L);

        return counterId;
    }
}

@@ -1206,31 +1206,29 @@
 tomicBuffer 
-counter
+value
 sBuffer;%0A   
@@ -1500,23 +1500,21 @@
  @param 
-counter
+value
 sBuffer 
@@ -1650,23 +1650,21 @@
 cBuffer 
-counter
+value
 sBuffer)
@@ -1725,23 +1725,21 @@
    this.
-counter
+value
 sBuffer 
@@ -1740,23 +1740,21 @@
 uffer = 
-counter
+value
 sBuffer;
@@ -1754,39 +1754,37 @@
 Buffer;%0A        
-counter
+value
 sBuffer.verifyAl
@@ -2048,17 +2048,21 @@
 terId = 
-c
+nextC
 ounterId
@@ -2180,23 +2180,21 @@
 NGTH) %3E 
-counter
+value
 sBuffer.
@@ -2734,23 +2734,21 @@
 Counter(
-counter
+value
 sBuffer,
@@ -4218,39 +4218,37 @@
 )%0A    %7B%0A        
-counter
+value
 sBuffer.putLongO
@@ -4417,25 +4417,29 @@
 private int 
-c
+nextC
 ounterId()%0A 
@@ -4588,23 +4588,21 @@
         
-counter
+value
 sBuffer.

Rename for consistency.