package io.aeron.driver;


import org.agrona.MutableDirectBuffer;
import org.agrona.concurrent.AtomicBuffer;
import org.agrona.concurrent.status.AtomicCounter;
import org.agrona.concurrent.status.CountersManager;

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

/**
 * A thread safe extension of the {@link CountersManager} to allow same process read and write access to the same
 * counter buffer. Note that cross process access is not catered for.
 *
 * TODO: move to Agrona when settled.
 */
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)
    {
        lock.lock();
        try
        {
            return super.allocate(label);
        }
        finally
        {
            lock.unlock();
        }
    }

    @Override
    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();
        }
    }

    @Override
    public AtomicCounter newCounter(final String label)
    {
        lock.lock();
        try
        {
            return super.newCounter(label);
        }
        finally
        {
            lock.unlock();
        }
    }

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

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

@@ -19,16 +19,47 @@
 river;%0A%0A
+import org.agrona.DirectBuffer;
 %0Aimport 
@@ -137,59 +137,8 @@
 er;%0A
-import org.agrona.concurrent.status.AtomicCounter;%0A
 impo
@@ -340,20 +340,16 @@
 sion of 
-the 
 %7B@link C
@@ -368,22 +368,27 @@
 er%7D 
-to
+which
  allow
- same 
+s intra-
 proc
@@ -435,16 +435,17 @@
  counter
+s
  buffer.
@@ -459,14 +459,14 @@
 hat 
-cross 
+inter-
 proc
@@ -1271,37 +1271,24 @@
    %7D%0A    %7D%0A%0A
-    @Override
 %0A    public 
@@ -1579,378 +1579,251 @@
     
-@Override%0A    public AtomicCounter newCounter(final String label)
+public int allocate(
 %0A    
-%7B%0A
     
-    lock.lock();%0A        try%0A        %7B%0A            return super.newCounter(label);%0A        %7D%0A
+final int typeId,%0A        final DirectBuffer keyBuffer,%0A  
       
-  
 final
-ly%0A        %7B%0A            lock.unlock();%0A        %7D%0A    %7D%0A%0A    @Override%0A    public AtomicCounter newCounter(
+ int keyOffset,%0A        final int keyLength,%0A        final DirectBuffer labelBuffer,%0A        
 final 
-Str
 in
-g
+t
  label
-, final int typeId, final Consumer%3CMutableDirectBuffer%3E keyFunc
+Offset,%0A        final int labelLength
 )%0A  
@@ -1898,41 +1898,95 @@
 per.
-newCounter(label, typeId, keyFunc
+allocate(typeId, keyBuffer, keyOffset, keyLength, labelBuffer, labelOffset, labelLength
 );%0A 

Update ConcurrentCountersManager for new allocated method.