// Path: src/org/jgroups/raft/RaftHandle.java
// public class RaftHandle implements Settable {
//     protected JChannel     ch;
//     protected RAFT         raft;
//     protected Settable     settable; // usually REDIRECT (at the top of the stack)
// 
//     /**
//      * Creates a RaftHandle instance.
//      * @param ch The channel over which to create the RaftHandle. Must be non-null, but doesn't yet need to be connected
//      * @param sm An implementation of {@link StateMachine}.
//      *           Can be null, ie. if it is set later via {@link #stateMachine(StateMachine)}.
//      */
//     public RaftHandle(JChannel ch, StateMachine sm) {
//         if((this.ch=ch) == null)
//             throw new IllegalStateException("channel must not be null");
//         if((raft=RAFT.findProtocol(RAFT.class, ch.getProtocolStack().getTopProtocol(),true)) == null)
//             throw new IllegalStateException("RAFT protocol was not found");
//         if((settable=RAFT.findProtocol(Settable.class, ch.getProtocolStack().getTopProtocol(),true)) == null)
//             throw new IllegalStateException("did not find a protocol implementing Settable (e.g. REDIRECT or RAFT)");
//         stateMachine(sm);
//     }
// 
//     public JChannel     channel()                                    {return ch;}
//     public RAFT         raft()                                       {return raft;}
//     public String       raftId()                                     {return raft.raftId();}
//     public RaftHandle   raftId(String id)                            {raft.raftId(id); return this;}
//     public Address      leader()                                     {return raft.leader();}
//     public boolean      isLeader()                                   {return raft.isLeader();}
//     public StateMachine stateMachine()                               {return raft.stateMachine();}
//     public RaftHandle   stateMachine(StateMachine sm)                {raft.stateMachine(sm); return this;}
//     public RaftHandle   addRoleListener(RAFT.RoleChange listener)    {raft.addRoleListener(listener); return this;}
//     public RaftHandle   removeRoleListener(RAFT.RoleChange listener) {raft.remRoleListener(listener); return this;}
//     public int          currentTerm()                                {return raft.currentTerm();}
//     public int          lastApplied()                                {return raft.lastAppended();}
//     public int          commitIndex()                                {return raft.commitIndex();}
//     public void         snapshot() throws Exception                  {raft.snapshot();}
//     public Log          log()                                        {return raft.log();}
//     public int          logSize()                                    {return raft.logSize();}
// 
//     public void logEntries(ObjIntConsumer<LogEntry> func) {
//         raft.logEntries(func);
//     }
// 
//     @Override
//     public CompletableFuture<byte[]> setAsync(byte[] buf, int offset, int length) throws Exception {
//         return settable.setAsync(buf, offset, length);
//     }
// 
// 
// }


import org.jgroups.JChannel;
import org.jgroups.blocks.atomic.Counter;
import org.jgroups.protocols.raft.*;
import org.jgroups.raft.RaftHandle;
import org.jgroups.util.*;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
package org.jgroups.raft.blocks;



/**
 * Provides a consensus based distributed counter (similar to AtomicLong) which can be atomically updated across a cluster.
 * @author Bela Ban
 * @since  0.2
 */
public class CounterService implements StateMachine, RAFT.RoleChange {
    protected JChannel   ch;
    protected RaftHandle raft;
    protected long       repl_timeout=20000; // timeout (ms) to wait for a majority to ack a write

    /** If true, reads can return the local counter value directly. Else, reads have to go through the leader */
    protected boolean  allow_dirty_reads=true;

    // keys: counter names, values: counter values
    protected final Map<String,Long> counters=new HashMap<>();

    protected enum Command {create, delete, get, set, compareAndSet, incrementAndGet, decrementAndGet, addAndGet}


    public CounterService(JChannel ch) {
        setChannel(ch);
    }

    public void setChannel(JChannel ch) {
        this.ch=ch;
        this.raft=new RaftHandle(this.ch, this);
        raft.addRoleListener(this);
    }

    public void           addRoleChangeListener(RAFT.RoleChange listener)  {raft.addRoleListener(listener);}
    public long           replTimeout()                 {return repl_timeout;}
    public CounterService replTimeout(long timeout)     {this.repl_timeout=timeout; return this;}
    public boolean        allowDirtyReads()             {return allow_dirty_reads;}
    public CounterService allowDirtyReads(boolean flag) {allow_dirty_reads=flag; return this;}
    public int            lastApplied()                 {return raft.lastApplied();}
    public int            commitIndex()                 {return raft.commitIndex();}
    public void           snapshot() throws Exception   {raft.snapshot();}
    public int            logSize()                     {return raft.logSize();}
    public String         raftId()                      {return raft.raftId();}
    public CounterService raftId(String id)             {raft.raftId(id); return this;}


    /**
     * Returns an instance of the counter. This is local operation which never fails and don't create a record in the
     * log. For cluster-wide operation call getOrCreateCounter(name, initial_value).
     * @param name Name of the counter, different counters have to have different names
     * @return The counter instance
     */
    public Counter counter(String name) {
        return new CounterImpl(name, this);
    }


    /**
     * Returns an existing counter, or creates a new one if none exists. This is a cluster-wide operation which would
     * fail if no leader is elected.
     * @param name Name of the counter, different counters have to have different names
     * @param initial_value The initial value of a new counter if there is no existing counter. Ignored
     * if the counter already exists
     * @return The counter implementation
     */
    public Counter getOrCreateCounter(String name, long initial_value) throws Exception {
        invoke(Command.create, name, false, initial_value);
        return new CounterImpl(name, this);
    }

  
    /**
     * Deletes a counter instance (on the coordinator)
     * @param name The name of the counter. No-op if the counter doesn't exist
     */
// Path: src/org/jgroups/raft/RaftHandle.java
// public class RaftHandle implements Settable {
//     protected JChannel     ch;
//     protected RAFT         raft;
//     protected Settable     settable; // usually REDIRECT (at the top of the stack)
// 
//     /**
//      * Creates a RaftHandle instance.
//      * @param ch The channel over which to create the RaftHandle. Must be non-null, but doesn't yet need to be connected
//      * @param sm An implementation of {@link StateMachine}.
//      *           Can be null, ie. if it is set later via {@link #stateMachine(StateMachine)}.
//      */
//     public RaftHandle(JChannel ch, StateMachine sm) {
//         if((this.ch=ch) == null)
//             throw new IllegalStateException("channel must not be null");
//         if((raft=RAFT.findProtocol(RAFT.class, ch.getProtocolStack().getTopProtocol(),true)) == null)
//             throw new IllegalStateException("RAFT protocol was not found");
//         if((settable=RAFT.findProtocol(Settable.class, ch.getProtocolStack().getTopProtocol(),true)) == null)
//             throw new IllegalStateException("did not find a protocol implementing Settable (e.g. REDIRECT or RAFT)");
//         stateMachine(sm);
//     }
// 
//     public JChannel     channel()                                    {return ch;}
//     public RAFT         raft()                                       {return raft;}
//     public String       raftId()                                     {return raft.raftId();}
//     public RaftHandle   raftId(String id)                            {raft.raftId(id); return this;}
//     public Address      leader()                                     {return raft.leader();}
//     public boolean      isLeader()                                   {return raft.isLeader();}
//     public StateMachine stateMachine()                               {return raft.stateMachine();}
//     public RaftHandle   stateMachine(StateMachine sm)                {raft.stateMachine(sm); return this;}
//     public RaftHandle   addRoleListener(RAFT.RoleChange listener)    {raft.addRoleListener(listener); return this;}
//     public RaftHandle   removeRoleListener(RAFT.RoleChange listener) {raft.remRoleListener(listener); return this;}
//     public int          currentTerm()                                {return raft.currentTerm();}
//     public int          lastApplied()                                {return raft.lastAppended();}
//     public int          commitIndex()                                {return raft.commitIndex();}
//     public void         snapshot() throws Exception                  {raft.snapshot();}
//     public Log          log()                                        {return raft.log();}
//     public int          logSize()                                    {return raft.logSize();}
// 
//     public void logEntries(ObjIntConsumer<LogEntry> func) {
//         raft.logEntries(func);
//     }
// 
//     @Override
//     public CompletableFuture<byte[]> setAsync(byte[] buf, int offset, int length) throws Exception {
//         return settable.setAsync(buf, offset, length);
//     }
// 
// 
// }


// Path: src/org/jgroups/raft/blocks/CounterService.java
import org.jgroups.JChannel;
import org.jgroups.blocks.atomic.Counter;
import org.jgroups.protocols.raft.*;
import org.jgroups.raft.RaftHandle;
import org.jgroups.util.*;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

package org.jgroups.raft.blocks;



/**
 * Provides a consensus based distributed counter (similar to AtomicLong) which can be atomically updated across a cluster.
 * @author Bela Ban
 * @since  0.2
 */
public class CounterService implements StateMachine, RAFT.RoleChange {
    protected JChannel   ch;
    protected RaftHandle raft;
    protected long       repl_timeout=20000; // timeout (ms) to wait for a majority to ack a write

    /** If true, reads can return the local counter value directly. Else, reads have to go through the leader */
    protected boolean  allow_dirty_reads=true;

    // keys: counter names, values: counter values
    protected final Map<String,Long> counters=new HashMap<>();

    protected enum Command {create, delete, get, set, compareAndSet, incrementAndGet, decrementAndGet, addAndGet}


    public CounterService(JChannel ch) {
        setChannel(ch);
    }

    public void setChannel(JChannel ch) {
        this.ch=ch;
        this.raft=new RaftHandle(this.ch, this);
        raft.addRoleListener(this);
    }

    public void           addRoleChangeListener(RAFT.RoleChange listener)  {raft.addRoleListener(listener);}
    public long           replTimeout()                 {return repl_timeout;}
    public CounterService replTimeout(long timeout)     {this.repl_timeout=timeout; return this;}
    public boolean        allowDirtyReads()             {return allow_dirty_reads;}
    public CounterService allowDirtyReads(boolean flag) {allow_dirty_reads=flag; return this;}
    public int            lastApplied()                 {return raft.lastApplied();}
    public int            commitIndex()                 {return raft.commitIndex();}
    public void           snapshot() throws Exception   {raft.snapshot();}
    public int            logSize()                     {return raft.logSize();}
    public String         raftId()                      {return raft.raftId();}
    public CounterService raftId(String id)             {raft.raftId(id); return this;}


    /**
     * Returns an instance of the counter. This is local operation which never fails and don't create a record in the
     * log. For cluster-wide operation call getOrCreateCounter(name, initial_value).
     * @param name Name of the counter, different counters have to have different names
     * @return The counter instance
     */
    public Counter counter(String name) {
        return new CounterImpl(name, this);
    }


    /**
     * Returns an existing counter, or creates a new one if none exists. This is a cluster-wide operation which would
     * fail if no leader is elected.
     * @param name Name of the counter, different counters have to have different names
     * @param initial_value The initial value of a new counter if there is no existing counter. Ignored
     * if the counter already exists
     * @return The counter implementation
     */
    public Counter getOrCreateCounter(String name, long initial_value) throws Exception {
        invoke(Command.create, name, false, initial_value);
        return new CounterImpl(name, this);
    }

  
    /**
     * Deletes a counter instance (on the coordinator)
     * @param name The name of the counter. No-op if the counter doesn't exist
     */
    public void deleteCounter(String name) throws Exception {