/*
 * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
 * and is available at http://www.eclipse.org/legal/epl-v10.html
 */
package org.opendaylight.controller.cluster.raft;

import akka.japi.Procedure;
import java.util.Collections;
import java.util.List;
import org.opendaylight.controller.cluster.raft.base.messages.DeleteEntries;
import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;

/**
 * Implementation of ReplicatedLog used by the RaftActor.
 */
class ReplicatedLogImpl extends AbstractReplicatedLogImpl {
    private static final int DATA_SIZE_DIVIDER = 5;

    private long dataSizeSinceLastSnapshot = 0L;
    private final RaftActorContext context;
    private final RaftActorBehavior currentBehavior;

    private final Procedure<DeleteEntries> deleteProcedure = new Procedure<DeleteEntries>() {
        @Override
        public void apply(DeleteEntries notUsed) {
        }
    };

    static ReplicatedLog newInstance(Snapshot snapshot, RaftActorContext context,
            RaftActorBehavior currentBehavior) {
        return new ReplicatedLogImpl(snapshot.getLastAppliedIndex(), snapshot.getLastAppliedTerm(),
                snapshot.getUnAppliedEntries(), context, currentBehavior);
    }

    static ReplicatedLog newInstance(RaftActorContext context, RaftActorBehavior currentBehavior) {
        return new ReplicatedLogImpl(-1L, -1L, Collections.<ReplicatedLogEntry>emptyList(), context,
                currentBehavior);
    }

    private ReplicatedLogImpl(long snapshotIndex, long snapshotTerm, List<ReplicatedLogEntry> unAppliedEntries,
            RaftActorContext context, RaftActorBehavior currentBehavior) {
        super(snapshotIndex, snapshotTerm, unAppliedEntries);
        this.context = context;
        this.currentBehavior = currentBehavior;
    }

    @Override
    public void removeFromAndPersist(long logEntryIndex) {
        // FIXME: Maybe this should be done after the command is saved
        long adjustedIndex = removeFrom(logEntryIndex);
        if(adjustedIndex >= 0) {
            context.getPersistenceProvider().persist(new DeleteEntries(adjustedIndex), deleteProcedure);
        }
    }

    @Override
    public void appendAndPersist(final ReplicatedLogEntry replicatedLogEntry) {
        appendAndPersist(replicatedLogEntry, null);
    }

    @Override
    public void captureSnapshotIfReady(ReplicatedLogEntry replicatedLogEntry) {
        long journalSize = replicatedLogEntry.getIndex() + 1;
        long dataThreshold = context.getTotalMemory() *
                context.getConfigParams().getSnapshotDataThresholdPercentage() / 100;

        if ((journalSize % context.getConfigParams().getSnapshotBatchCount() == 0
                || getDataSizeForSnapshotCheck() > dataThreshold)) {

            boolean started = context.getSnapshotManager().capture(replicatedLogEntry,
                    currentBehavior.getReplicatedToAllIndex());
            if (started) {
                if (!context.hasFollowers()) {
                    dataSizeSinceLastSnapshot = 0;
                }
            }
        }
    }

    private long getDataSizeForSnapshotCheck() {
        long dataSizeForCheck = dataSize();
        if (!context.hasFollowers()) {
            // When we do not have followers we do not maintain an in-memory log
            // due to this the journalSize will never become anything close to the
            // snapshot batch count. In fact will mostly be 1.
            // Similarly since the journal's dataSize depends on the entries in the
            // journal the journal's dataSize will never reach a value close to the
            // memory threshold.
            // By maintaining the dataSize outside the journal we are tracking essentially
            // what we have written to the disk however since we no longer are in
            // need of doing a snapshot just for the sake of freeing up memory we adjust
            // the real size of data by the DATA_SIZE_DIVIDER so that we do not snapshot as often
            // as if we were maintaining a real snapshot
            dataSizeForCheck = dataSizeSinceLastSnapshot / DATA_SIZE_DIVIDER;
        }
        return dataSizeForCheck;
    }

    @Override
    public void appendAndPersist(final ReplicatedLogEntry replicatedLogEntry,
            final Procedure<ReplicatedLogEntry> callback)  {

        if(context.getLogger().isDebugEnabled()) {
            context.getLogger().debug("{}: Append log entry and persist {} ", context.getId(), replicatedLogEntry);
        }

        // FIXME : By adding the replicated log entry to the in-memory journal we are not truly ensuring durability of the logs
        append(replicatedLogEntry);

        // When persisting events with persist it is guaranteed that the
        // persistent actor will not receive further commands between the
        // persist call and the execution(s) of the associated event
        // handler. This also holds for multiple persist calls in context
        // of a single command.
        context.getPersistenceProvider().persist(replicatedLogEntry,
            new Procedure<ReplicatedLogEntry>() {
                @Override
                public void apply(ReplicatedLogEntry evt) throws Exception {
                    context.getLogger().debug("{}: persist complete {}", context.getId(), replicatedLogEntry);

                    int logEntrySize = replicatedLogEntry.size();
                    dataSizeSinceLastSnapshot += logEntrySize;

                    if (callback != null){
                        callback.apply(replicatedLogEntry);
                    }
                }
            }
        );
    }
}
@@ -380,16 +380,61 @@
 cedure;%0A
+import com.google.common.base.Preconditions;%0A
 import j
@@ -1097,16 +1097,22 @@
 d apply(
+final 
 DeleteEn
@@ -1183,16 +1183,22 @@
 nstance(
+final 
 Snapshot
@@ -1207,16 +1207,22 @@
 napshot,
+ final
  RaftAct
@@ -1244,32 +1244,38 @@
 xt,%0A            
+final 
 RaftActorBehavio
@@ -1514,16 +1514,22 @@
 nstance(
+final 
 RaftActo
@@ -1537,32 +1537,38 @@
 Context context,
+ final
  RaftActorBehavi
@@ -1761,16 +1761,22 @@
 LogImpl(
+final 
 long sna
@@ -1786,16 +1786,22 @@
 otIndex,
+ final
  long sn
@@ -1811,16 +1811,22 @@
 hotTerm,
+ final
  List%3CRe
@@ -1873,16 +1873,22 @@
         
+final 
 RaftActo
@@ -1904,16 +1904,22 @@
 context,
+ final
  RaftAct
@@ -2033,23 +2033,51 @@
 ntext = 
+Preconditions.checkNotNull(
 context
+)
 ;%0A      
@@ -2101,16 +2101,43 @@
 avior = 
+Preconditions.checkNotNull(
 currentB
@@ -2143,16 +2143,17 @@
 Behavior
+)
 ;%0A    %7D%0A
@@ -2204,16 +2204,22 @@
 Persist(
+final 
 long log
@@ -2720,16 +2720,22 @@
 IfReady(
+final 
 Replicat
@@ -4718,16 +4718,17 @@
       if
+ 
 (context
@@ -5553,16 +5553,22 @@
 d apply(
+final 
 Replicat
@@ -5582,11 +5582,13 @@
 try 
-evt
+param
 ) th
@@ -5694,34 +5694,21 @@
 etId(), 
-replicatedLogEntry
+param
 );%0A%0A    
@@ -5742,34 +5742,21 @@
 ySize = 
-replicatedLogEntry
+param
 .size();
@@ -5861,16 +5861,17 @@
 != null)
+ 
 %7B%0A      
@@ -5903,34 +5903,21 @@
 k.apply(
-replicatedLogEntry
+param
 );%0A     

Use local param in persist callback