/*
 * Copyright 2017 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 org.agrona.concurrent.status;

import org.agrona.DirectBuffer;
import org.agrona.MutableDirectBuffer;
import org.agrona.concurrent.AtomicBuffer;

import java.nio.charset.Charset;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;

/**
 * A thread safe extension of {@link CountersManager} which allows intra-process read and write access to the same
 * counters buffer. Note that inter-process access is not catered for.
 */
public class ConcurrentCountersManager extends CountersManager
{
    private final ReentrantLock lock = new ReentrantLock();

    public ConcurrentCountersManager(final AtomicBuffer metaDataBuffer, final AtomicBuffer valuesBuffer)
    {
        super(metaDataBuffer, valuesBuffer);
    }

    public ConcurrentCountersManager(
        final AtomicBuffer metaDataBuffer,
        final AtomicBuffer valuesBuffer,
        final Charset labelCharset)
    {
        super(metaDataBuffer, valuesBuffer, labelCharset);
    }

    public int allocate(final String label, final int typeId)
    {
        lock.lock();
        try
        {
            return super.allocate(label, typeId);
        }
        finally
        {
            lock.unlock();
        }
    }


    public int allocate(final String label, final int typeId, final Consumer<MutableDirectBuffer> keyFunc)
    {
        lock.lock();
        try
        {
            return super.allocate(label, typeId, keyFunc);
        }
        finally
        {
            lock.unlock();
        }
    }

    public int allocate(
        final int typeId,
        final DirectBuffer keyBuffer,
        final int keyOffset,
        final int keyLength,
        final DirectBuffer labelBuffer,
        final int labelOffset,
        final int labelLength)
    {
        lock.lock();
        try
        {
            return super.allocate(typeId, keyBuffer, keyOffset, keyLength, labelBuffer, labelOffset, labelLength);
        }
        finally
        {
            lock.unlock();
        }
    }

    public void free(final int counterId)
    {
        lock.lock();
        try
        {
            super.free(counterId);
        }
        finally
        {
            lock.unlock();
        }
    }
}

@@ -740,16 +740,57 @@
 cBuffer;
+%0Aimport org.agrona.concurrent.EpochClock;
 %0A%0Aimport
@@ -1607,32 +1607,374 @@
 harset);%0A    %7D%0A%0A
+    public ConcurrentCountersManager(%0A        final AtomicBuffer metaDataBuffer,%0A        final AtomicBuffer valuesBuffer,%0A        final Charset labelCharset,%0A        final EpochClock epochClock,%0A        final long freeToReuseTimeoutMs)%0A    %7B%0A        super(metaDataBuffer, valuesBuffer, labelCharset, epochClock, freeToReuseTimeoutMs);%0A    %7D%0A%0A
     public int a
@@ -2015,24 +2015,24 @@
 int typeId)%0A
-
     %7B%0A      
@@ -2195,16 +2195,15 @@
   %7D%0A
+
     %7D%0A%0A
-%0A
     

Support free to reuse counter deadline in ConcurrentCountersManager.