@Override
public void save(final SnapshotWriter writer, final Region region, final Closure done,
                 final ExecutorService executor) {
    final String writerPath = writer.getPath();
    final String snapshotPath = Paths.get(writerPath, SNAPSHOT_DIR).toString();
    try {
        doSnapshotSave(snapshotPath, region, executor).whenComplete((metaBuilder, throwable) -> {
            if (throwable == null) {
                executor.execute(() -> compressSnapshot(writer, metaBuilder, done));
            } else {
                LOG.error("Fail to save snapshot, path={}, file list={}, {}.", writerPath, writer.listFiles(),
                        StackTraceUtil.stackTrace(throwable));
                done.run(new Status(RaftError.EIO, "Fail to save snapshot at %s, error is %s", writerPath,
                        throwable.getMessage()));
            }
        });
    } catch (final Throwable t) {
        LOG.error("Fail to save snapshot, path={}, file list={}, {}.", writerPath, writer.listFiles(),
                StackTraceUtil.stackTrace(t));
        done.run(new Status(RaftError.EIO, "Fail to save snapshot at %s, error is %s", writerPath,
                t.getMessage()));
    }
}
 

protected void compressSnapshot(final SnapshotWriter writer, final LocalFileMeta.Builder metaBuilder,
                                final Closure done) {
    final String writerPath = writer.getPath();
    final String outputFile = Paths.get(writerPath, SNAPSHOT_ARCHIVE).toString();
    try {
        final Checksum checksum = new CRC64();
        ZipUtil.compress(writerPath, SNAPSHOT_DIR, outputFile, checksum);
        metaBuilder.setChecksum(Long.toHexString(checksum.getValue()));
        if (writer.addFile(SNAPSHOT_ARCHIVE, metaBuilder.build())) {
            done.run(Status.OK());
        } else {
            done.run(new Status(RaftError.EIO, "Fail to add snapshot file: %s", writerPath));
        }
    } catch (final Throwable t) {
        LOG.error("Fail to compress snapshot, path={}, file list={}, {}.", writerPath, writer.listFiles(),
            StackTraceUtil.stackTrace(t));
        done.run(new Status(RaftError.EIO, "Fail to compress snapshot at %s, error is %s", writerPath, t
            .getMessage()));
    }
}
 

@Override
public void onSnapshotSave(final SnapshotWriter writer, final Closure done) {
  final Map<String, Long> values = new HashMap<>();
  for (final Map.Entry<String, AtomicLong> entry : this.counters.entrySet()) {
    values.put(entry.getKey(), entry.getValue().get());
  }
  Utils.runInThread(() -> {
    final AtomicSnapshotFile snapshot = new AtomicSnapshotFile(writer.getPath() + File.separator + "data");
    if (snapshot.save(values)) {
      if (writer.addFile("data")) {
        done.run(Status.OK());
      } else {
        done.run(new Status(RaftError.EIO, "Fail to add file to writer"));
      }
    } else {
      done.run(new Status(RaftError.EIO, "Fail to save counter snapshot %s", snapshot.getPath()));
    }
  });
}
 

@Override
public void clear() {
    List<Closure> savedQueue;
    this.lock.lock();
    try {
        this.firstIndex = 0;
        savedQueue = this.queue;
        this.queue = new LinkedList<>();
    } finally {
        this.lock.unlock();
    }

    final Status status = new Status(RaftError.EPERM, "Leader stepped down");
    Utils.runInThread(() -> {
        for (final Closure done : savedQueue) {
            if (done != null) {
                done.run(status);
            }
        }
    });
}
 

/**
 * Called by leader, otherwise the behavior is undefined
 * Store application context before replication.
 *
 * @param conf      current configuration
 * @param oldConf   old configuration
 * @param done      callback
 * @return          returns true on success
 */
public boolean appendPendingTask(final Configuration conf, final Configuration oldConf, final Closure done) {
    final Ballot bl = new Ballot();
    if (!bl.init(conf, oldConf)) {
        LOG.error("Fail to init ballot.");
        return false;
    }
    final long stamp = this.stampedLock.writeLock();
    try {
        if (this.pendingIndex <= 0) {
            LOG.error("Fail to appendingTask, pendingIndex={}.", this.pendingIndex);
            return false;
        }
        this.pendingMetaQueue.add(bl);
        this.closureQueue.appendPendingClosure(done);
        return true;
    } finally {
        this.stampedLock.unlockWrite(stamp);
    }
}
 

private void afterShutdown() {
    List<Closure> savedDoneList = null;
    this.writeLock.lock();
    try {
        if (!this.shutdownContinuations.isEmpty()) {
            savedDoneList = new ArrayList<>(this.shutdownContinuations);
        }
        if (this.logStorage != null) {
            this.logStorage.shutdown();
        }
        this.state = State.STATE_SHUTDOWN;
    } finally {
        this.writeLock.unlock();
    }
    if (savedDoneList != null) {
        for (final Closure closure : savedDoneList) {
            Utils.runClosureInThread(closure);
        }
    }
}
 

@Test
public void testHandleRequest() {
    this.mockNodes(3);
    Mockito.when(this.node.getGroupId()).thenReturn(this.groupId);
    PeerId peerId = new PeerId();
    peerId.parse(this.peerIdStr);
    Mockito.when(this.node.getOptions()).thenReturn(new NodeOptions());
    Mockito.when(this.node.getNodeId()).thenReturn(new NodeId("test", peerId));
    NodeManager.getInstance().addAddress(peerId.getEndpoint());
    NodeManager.getInstance().add(this.node);

    BaseCliRequestProcessor<T> processor = newProcessor();
    processor.handleRequest(this.asyncContext, createRequest(this.groupId, peerId));
    ArgumentCaptor<Closure> doneArg = ArgumentCaptor.forClass(Closure.class);
    verify(processor.interest(), this.node, doneArg);
}
 

@Test
public void testResetFirstIndex() {
    assertEquals(0, this.queue.getFirstIndex());
    this.queue.resetFirstIndex(10);
    assertEquals(10, this.queue.getFirstIndex());
    for (int i = 0; i < 10; i++) {
        this.queue.appendPendingClosure(mockClosure(null));
    }

    List<Closure> closures = new ArrayList<>();
    assertEquals(5, this.queue.popClosureUntil(4, closures));
    assertTrue(closures.isEmpty());
    assertEquals(4, this.queue.popClosureUntil(3, closures));
    assertTrue(closures.isEmpty());

    assertEquals(10, this.queue.popClosureUntil(19, closures));
    assertEquals(20, this.queue.getFirstIndex());
    assertEquals(10, closures.size());
    // empty ,return index+1
    assertEquals(21, this.queue.popClosureUntil(20, closures));
    assertTrue(closures.isEmpty());
}
 

@Override
public void onSnapshotSave(final SnapshotWriter writer, final Closure done) {
  final long currVal = this.value.get();
  Utils.runInThread(() -> {
    final CounterSnapshotFile snapshot = new CounterSnapshotFile(writer.getPath() + File.separator + "data");
    if (snapshot.save(currVal)) {
      if (writer.addFile("data")) {
        done.run(Status.OK());
      } else {
        done.run(new Status(RaftError.EIO, "Fail to add file to writer"));
      }
    } else {
      done.run(new Status(RaftError.EIO, "Fail to save counter snapshot %s", snapshot.getPath()));
    }
  });
}
 

@Override
public void onSnapshotSave(final SnapshotWriter writer, final Closure done) {
    this.saveSnapshotTimes++;
    final String path = writer.getPath() + File.separator + "data";
    final File file = new File(path);
    try (FileOutputStream fout = new FileOutputStream(file);
            BufferedOutputStream out = new BufferedOutputStream(fout)) {
        this.lock.lock();
        try {
            for (final ByteBuffer buf : this.logs) {
                final byte[] bs = new byte[4];
                Bits.putInt(bs, 0, buf.remaining());
                out.write(bs);
                out.write(buf.array());
            }
            this.snapshotIndex = this.appliedIndex;
        } finally {
            this.lock.unlock();
        }
        System.out.println("Node<" + this.address + "> saved snapshot into " + file);
        writer.addFile("data");
        done.run(Status.OK());
    } catch (final IOException e) {
        e.printStackTrace();
        done.run(new Status(RaftError.EIO, "Fail to save snapshot"));
    }
}
 

/**
 * Creates a task with data/done/expectedTerm.
 */
public Task(ByteBuffer data, Closure done, long expectedTerm) {
    super();
    this.data = data;
    this.done = done;
    this.expectedTerm = expectedTerm;
}
 

/**
 * Waiting for all tasks to complete with a timeout millis.
 *
 * @param tasks         task list
 * @param timeoutMillis the maximum millis to wait
 * @return the closure list in the tasks
 * @throws InterruptedException if the current thread is interrupted while waiting
 * @throws TimeoutException if timeout
 * @since 1.3.1
 */
public static List<Closure> joinAll(final List<Task> tasks, long timeoutMillis) throws InterruptedException,
                                                                               TimeoutException {
    final List<Closure> closures = new ArrayList<>(tasks.size());
    for (final Task t : tasks) {
        final long start = System.nanoTime();
        closures.add(t.join(timeoutMillis));
        timeoutMillis -= TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
        if (timeoutMillis <= 0) {
            throw new TimeoutException("joined timeout");
        }
    }
    return closures;
}
 

/**
 * Run closure with OK status in thread pool.
 */
public static Future<?> runClosureInThread(final Closure done) {
    if (done == null) {
        return null;
    }
    return runClosureInThread(done, Status.OK());
}
 

/**
 * Run closure with status in thread pool.
 */
public static Future<?> runClosureInThread(final Closure done, final Status status) {
    if (done == null) {
        return null;
    }

    return runInThread(() -> {
        try {
            done.run(status);
        } catch (final Throwable t) {
            LOG.error("Fail to run done closure", t);
        }
    });
}
 

/**
 * Run closure with OK status in thread pool.
 */
public static Future<?> runClosureInThread(final Closure done) {
    if (done == null) {
        return null;
    }
    return runClosureInThread(done, Status.OK());
}
 

/**
 * Run closure with status in thread pool.
 */
public static Future<?> runClosureInThread(final Closure done, final Status status) {
    if (done == null) {
        return null;
    }
    return runInThread(() -> {
        try {
            done.run(status);
        } catch (final Throwable t) {
            LOG.error("Fail to run done closure", t);
        }
    });
}
 

@Override
public void appendPendingClosure(final Closure closure) {
    this.lock.lock();
    try {
        this.queue.add(closure);
    } finally {
        this.lock.unlock();
    }
}
 

@Override
public long popClosureUntil(final long endIndex, final List<Closure> closures, final List<TaskClosure> taskClosures) {
    closures.clear();
    if (taskClosures != null) {
        taskClosures.clear();
    }
    this.lock.lock();
    try {
        final int queueSize = this.queue.size();
        if (queueSize == 0 || endIndex < this.firstIndex) {
            return endIndex + 1;
        }
        if (endIndex > this.firstIndex + queueSize - 1) {
            LOG.error("Invalid endIndex={}, firstIndex={}, closureQueueSize={}", endIndex, this.firstIndex,
                queueSize);
            return -1;
        }
        final long outFirstIndex = this.firstIndex;
        for (long i = outFirstIndex; i <= endIndex; i++) {
            final Closure closure = this.queue.pollFirst();
            if (taskClosures != null && closure instanceof TaskClosure) {
                taskClosures.add((TaskClosure) closure);
            }
            closures.add(closure);
        }
        this.firstIndex = endIndex + 1;
        return outFirstIndex;
    } finally {
        this.lock.unlock();
    }
}
 

private boolean passByStatus(final Closure done) {
    final Status status = this.error.getStatus();
    if (!status.isOk()) {
        if (done != null) {
            done.run(new Status(RaftError.EINVAL, "FSMCaller is in bad status=`%s`", status));
            return false;
        }
    }
    return true;
}
 

public IteratorImpl(final StateMachine fsm, final LogManager logManager, final List<Closure> closures,
                    final long firstClosureIndex, final long lastAppliedIndex, final long committedIndex,
                    final AtomicLong applyingIndex) {
    super();
    this.fsm = fsm;
    this.logManager = logManager;
    this.closures = closures;
    this.firstClosureIndex = firstClosureIndex;
    this.currentIndex = lastAppliedIndex;
    this.committedIndex = committedIndex;
    this.applyingIndex = applyingIndex;
    next();
}
 

protected void runTheRestClosureWithError() {
    for (long i = Math.max(this.currentIndex, this.firstClosureIndex); i <= this.committedIndex; i++) {
        final Closure done = this.closures.get((int) (i - this.firstClosureIndex));
        if (done != null) {
            Requires.requireNonNull(this.error, "error");
            Requires.requireNonNull(this.error.getStatus(), "error.status");
            final Status status = this.error.getStatus();
            Utils.runClosureInThread(done, status);
        }
    }
}
 

/**
 * Start change configuration.
 */
void start(final Configuration oldConf, final Configuration newConf, final Closure done) {
    if (isBusy()) {
        if (done != null) {
            Utils.runClosureInThread(done, new Status(RaftError.EBUSY, "Already in busy stage."));
        }
        throw new IllegalStateException("Busy stage");
    }
    if (this.done != null) {
        if (done != null) {
            Utils.runClosureInThread(done, new Status(RaftError.EINVAL, "Already have done closure."));
        }
        throw new IllegalArgumentException("Already have done closure");
    }
    this.done = done;
    this.stage = Stage.STAGE_CATCHING_UP;
    this.oldPeers = oldConf.listPeers();
    this.newPeers = newConf.listPeers();
    this.oldLearners = oldConf.listLearners();
    this.newLearners = newConf.listLearners();
    final Configuration adding = new Configuration();
    final Configuration removing = new Configuration();
    newConf.diff(oldConf, adding, removing);
    this.nchanges = adding.size() + removing.size();

    addNewLearners();
    if (adding.isEmpty()) {
        nextStage();
        return;
    }
    addNewPeers(adding);
}
 

private void unsafeRegisterConfChange(final Configuration oldConf, final Configuration newConf, final Closure done) {

        Requires.requireTrue(newConf.isValid(), "Invalid new conf: %s", newConf);
        // The new conf entry(will be stored in log manager) should be valid
        Requires.requireTrue(new ConfigurationEntry(null, newConf, oldConf).isValid(), "Invalid conf entry: %s",
            newConf);

        if (this.state != State.STATE_LEADER) {
            LOG.warn("Node {} refused configuration changing as the state={}.", getNodeId(), this.state);
            if (done != null) {
                final Status status = new Status();
                if (this.state == State.STATE_TRANSFERRING) {
                    status.setError(RaftError.EBUSY, "Is transferring leadership.");
                } else {
                    status.setError(RaftError.EPERM, "Not leader");
                }
                Utils.runClosureInThread(done, status);
            }
            return;
        }
        // check concurrent conf change
        if (this.confCtx.isBusy()) {
            LOG.warn("Node {} refused configuration concurrent changing.", getNodeId());
            if (done != null) {
                Utils.runClosureInThread(done, new Status(RaftError.EBUSY, "Doing another configuration change."));
            }
            return;
        }
        // Return immediately when the new peers equals to current configuration
        if (this.conf.getConf().equals(newConf)) {
            Utils.runClosureInThread(done);
            return;
        }
        this.confCtx.start(oldConf, newConf, done);
    }
 

@Override
public void addPeer(final PeerId peer, final Closure done) {
    Requires.requireNonNull(peer, "Null peer");
    this.writeLock.lock();
    try {
        Requires.requireTrue(!this.conf.getConf().contains(peer), "Peer already exists in current configuration");

        final Configuration newConf = new Configuration(this.conf.getConf());
        newConf.addPeer(peer);
        unsafeRegisterConfChange(this.conf.getConf(), newConf, done);
    } finally {
        this.writeLock.unlock();
    }
}
 

@Override
public void removePeer(final PeerId peer, final Closure done) {
    Requires.requireNonNull(peer, "Null peer");
    this.writeLock.lock();
    try {
        Requires.requireTrue(this.conf.getConf().contains(peer), "Peer not found in current configuration");

        final Configuration newConf = new Configuration(this.conf.getConf());
        newConf.removePeer(peer);
        unsafeRegisterConfChange(this.conf.getConf(), newConf, done);
    } finally {
        this.writeLock.unlock();
    }
}
 

@Override
public void changePeers(final Configuration newPeers, final Closure done) {
    Requires.requireNonNull(newPeers, "Null new peers");
    Requires.requireTrue(!newPeers.isEmpty(), "Empty new peers");
    this.writeLock.lock();
    try {
        LOG.info("Node {} change peers from {} to {}.", getNodeId(), this.conf.getConf(), newPeers);
        unsafeRegisterConfChange(this.conf.getConf(), newPeers, done);
    } finally {
        this.writeLock.unlock();
    }
}
 

@Override
public void addLearners(final List<PeerId> learners, final Closure done) {
    checkPeers(learners);
    this.writeLock.lock();
    try {
        final Configuration newConf = new Configuration(this.conf.getConf());
        for (final PeerId peer : learners) {
            newConf.addLearner(peer);
        }
        unsafeRegisterConfChange(this.conf.getConf(), newConf, done);
    } finally {
        this.writeLock.unlock();
    }

}
 

@Override
public void removeLearners(final List<PeerId> learners, final Closure done) {
    checkPeers(learners);
    this.writeLock.lock();
    try {
        final Configuration newConf = new Configuration(this.conf.getConf());
        for (final PeerId peer : learners) {
            newConf.removeLearner(peer);
        }
        unsafeRegisterConfChange(this.conf.getConf(), newConf, done);
    } finally {
        this.writeLock.unlock();
    }
}
 

@Override
public void resetLearners(final List<PeerId> learners, final Closure done) {
    checkPeers(learners);
    this.writeLock.lock();
    try {
        final Configuration newConf = new Configuration(this.conf.getConf());
        newConf.setLearners(new LinkedHashSet<>(learners));
        unsafeRegisterConfChange(this.conf.getConf(), newConf, done);
    } finally {
        this.writeLock.unlock();
    }
}
 

private void doSnapshot(final Closure done) {
    if (this.snapshotExecutor != null) {
        this.snapshotExecutor.doSnapshot(done);
    } else {
        if (done != null) {
            final Status status = new Status(RaftError.EINVAL, "Snapshot is not supported");
            Utils.runClosureInThread(done, status);
        }
    }
}