func (clog *CommitLog) LastLogEntryInfo() (lastIndex int64, lastTerm int64, err error) {
	lastIndex, err = clog.log.GetLastIndex()
	if lastIndex == 0 {
		return 0, 0, nil
	}
	msg, err := clog.log.GetMessages(lastIndex, 1)
	if err != nil {
		clog.node_log("Error retrieveing message with last index %v\n:", err)
		return lastIndex, 0, err
	}
	if msg.GetCount() < 1 {
		clog.node_log("Unable to retrieve message %v for last long entry info.\n", lastIndex)
	}
	msgTerm, err := msg.GetMessageTerm(0)
	if err != nil {
		clog.node_log("Error getting term from message with last index %v\n:", err)
		return lastIndex, 0, err
	}
	return lastIndex, msgTerm, nil
}
Used for voting - returns the last term and index in the log
package commitlog
import (
	"errors"
	"github.com/owlfish/forestbus-server/model"
	"github.com/owlfish/forestbus-server/utils"
	"sync"
	"time"
)
type LogStorage interface {
	GetMessages(index int64, count int64) (model.Messages, error)
	AppendMessages(msgs model.Messages) (lastIndex int64, err error)
	AppendFirstMessages(msgs model.Messages, leaderFirstIndex int64) (lastIndex int64, err error)
	Sync() error
	GetLastIndex() (lastIndex int64, err error)
	GetFirstIndex() (firstIndex int64, err error)
	TruncateMessages(index int64) error
	ExpVar() interface{}
	Shutdown(notifier *utils.ShutdownNotifier)
}
const MAX_MESSAGE_RESULTS = 500
const MAX_RETRIEVE_MESSAGE_COUNT = 2500
type DiskSyncPolicy int
const (
	WRITE_SYNC DiskSyncPolicy = iota
	PERIODIC_SYNC
	OSSYNC
)
const DEFAULT_SYNC_POLICY = OSSYNC
const DEFAULT_PERIODIC_SYNC_PERIOD = 5 * time.Second
type CommitLog struct {
	lock sync.RWMutex
	commitIndex int64
	waitingReaders *sync.Cond
	log LogStorage
	syncPolicy DiskSyncPolicy
	node_log utils.PrintFFunc
	inShutdown bool
}
type CommitLogInfo struct {
	LastCommittedID int64
	Storage         interface{}
}
func (clog *CommitLog) SetupLog(topicName string, store LogStorage) (err error) {
	clog.waitingReaders = sync.NewCond(clog.lock.RLocker())
	clog.node_log = utils.GetTopicLogger(topicName, "CommitLog")
	clog.inShutdown = false
	clog.log = store
	clog.commitIndex = 0
	clog.syncPolicy = DEFAULT_SYNC_POLICY
	if clog.syncPolicy == PERIODIC_SYNC {
		go func() {
			timer := time.NewTimer(DEFAULT_PERIODIC_SYNC_PERIOD)
			for {
				<-timer.C
				clog.log.Sync()
				timer.Reset(DEFAULT_PERIODIC_SYNC_PERIOD)
			}
		}()
	}
	return nil
}
func (clog *CommitLog) Append(msgs model.Messages, previousIndex int64, previousTerm int64, leaderFirstIndex int64) (nextIndex int64, previousMatch bool, err error) {
	overlapMessageCount := 0
	lastIndex, err := clog.log.GetLastIndex()
	nextIndex = lastIndex + 1
	if err != nil {
		clog.node_log("Error getting last message details during append: %v\n", err)
		return 0, false, err
	}
	if lastIndex > 0 {
		previousMessageList, err := clog.log.GetMessages(previousIndex, 1)
		if err != nil {
			return nextIndex, false, err
		}
		if previousMessageList.GetCount() < 1 {
			clog.node_log("Previous index given by leader (%v) is greater than or smaller than our last index (%v), returning no match\n", previousIndex, lastIndex)
			return nextIndex, false, nil
		}
		previousMessageTerm, err := previousMessageList.GetMessageTerm(0)
		if err != nil {
			clog.node_log("Error getting previous term: %v\n", err)
			return nextIndex, false, err
		}
		if previousMessageTerm != previousTerm {
			clog.node_log("Previous term of %v doesn't match previousMessageTerm of %v\n", previousTerm, previousMessageTerm)
			return nextIndex, false, nil
		}
		msgsToCheck := msgs.GetCount()
		if (int64(msgsToCheck) + previousIndex) >= lastIndex {
			msgsToCheck = int(lastIndex - previousIndex)
		}
		checkMessagesIndex := previousIndex
		if msgsToCheck > 0 {
			overlappingMessages, err := clog.getAtLeastMessages(checkMessagesIndex+1, int64(msgsToCheck))
			clog.node_log("Retrieved %v messages from index %v to check for overlaps (needed at least %v)\n", overlappingMessages.GetCount(), checkMessagesIndex+1, msgsToCheck)
			if err != nil {
				return nextIndex, false, err
			}
			for i := 0; i < overlappingMessages.GetCount(); i++ {
				checkMessagesIndex++
				overlappingMessagesTerm, err := overlappingMessages.GetMessageTerm(i)
				if err != nil {
					return nextIndex, false, err
				}
				msgsTerm, err := msgs.GetMessageTerm(overlapMessageCount)
				if err != nil {
					return nextIndex, false, err
				}
				if overlappingMessagesTerm != msgsTerm {
					clog.node_log("Truncating log due to mismatching terms to index %v.\n", checkMessagesIndex)
					clog.log.TruncateMessages(int64(checkMessagesIndex))
					break
				}
				overlapMessageCount++
			}
		}
	} else {
		if previousIndex != 0 {
			clog.node_log("Empty log, but previousIndex of %v given by leader\n", previousIndex)
			return nextIndex, false, nil
		}
		if msgs.GetCount() > 0 {
			lastID, err := clog.log.AppendFirstMessages(msgs, leaderFirstIndex)
			if err != nil {
				clog.node_log("Error attempting to Append messages to the log: %v\n", err)
				return nextIndex, false, nil
			} else if clog.syncPolicy == WRITE_SYNC {
				err = clog.log.Sync()
				if err != nil {
					clog.node_log("Error attempting to sync Append messages to the log: %v\n", err)
					return nextIndex, false, nil
				}
			}
			nextIndex = lastID + 1
			return nextIndex, true, nil
		}
	}
	msgsToAppend, err := msgs.Slice(overlapMessageCount, msgs.GetCount())
	if err != nil {
		clog.node_log("Error slicing overlapping messages: %v\n", err)
		return nextIndex, false, nil
	}
	if msgsToAppend.GetCount() > 0 {
		lastID, err := clog.log.AppendMessages(msgsToAppend)
		if err != nil {
			clog.node_log("Error attempting to Append messages to the log: %v\n", err)
			return nextIndex, false, nil
		} else if clog.syncPolicy == WRITE_SYNC {
			err = clog.log.Sync()
			if err != nil {
				clog.node_log("Error attempting to sync Append messages to the log: %v\n", err)
				return nextIndex, false, nil
			}
		}
		nextIndex = lastID + 1
	} else {
	}
	return nextIndex, true, nil
}
func (clog *CommitLog) Queue(term int64, modelMsgs model.Messages) (IDs []int64, err error) {
	if modelMsgs.GetCount() == 0 {
		clog.node_log("No messages found from client for Queue.")
		return nil, errors.New("No messages found to be Queued.")
	}
	err = modelMsgs.SetMessageTerm(term)
	if err != nil {
		clog.node_log("Error setting terms for messages: %v\n", err)
		return nil, err
	}
	lastIndex, err := clog.log.AppendMessages(modelMsgs)
	if err != nil {
		clog.node_log("Error writing message to log in Queue: %v\n", err)
		return nil, err
	} else if clog.syncPolicy == WRITE_SYNC {
		err = clog.log.Sync()
		if err != nil {
			clog.node_log("Error syncing written message to log in Queue: %v\n", err)
			return nil, err
		}
	}
	numMessages := modelMsgs.GetCount()
	results := make([]int64, numMessages)
	index := lastIndex - int64(numMessages) + 1
	for i := 0; i < numMessages; i++ {
		results[i] = index
		index++
	}
	return results, nil
}
func (clog *CommitLog) Get(ID int64, quantity int, waitForMessages bool) (msgs [][]byte, nextID int64, err error) {
	clog.lock.RLock()
	defer clog.lock.RUnlock()
	maxMessageResults := int64(quantity)
	if maxMessageResults > MAX_MESSAGE_RESULTS {
		maxMessageResults = MAX_MESSAGE_RESULTS
	}
	for !clog.inShutdown {
		if clog.commitIndex >= ID {
			messageCount := clog.commitIndex - ID + 1
			if messageCount > maxMessageResults {
				messageCount = maxMessageResults
			}
			modelMessages, err := clog.log.GetMessages(ID, messageCount)
			if err != nil {
				clog.node_log("Error getting messages: %v\n", err)
				return msgs, ID, err
			}
			msgs, err := modelMessages.Payloads()
			if err != nil {
				clog.node_log("Error getting payloads from messages: %v\n", err)
				return msgs, ID, err
			}
			if len(msgs) == 0 {
				ourFirstID, err := clog.log.GetFirstIndex()
				if err != nil {
					clog.node_log("Error getting messages during GetFirstIndex request: %v\n", err)
					return msgs, ID, err
				}
				if ourFirstID >= ID {
					nextID = ourFirstID
				} else {
					clog.node_log("Error getting messages for index %v which is higher than our first index of %v but brought back no messages\n", ID, ourFirstID)
					return nil, ID, errors.New("Error retrieveing messages even though index is higher than our first ID.")
				}
			} else {
				nextID = ID + int64(modelMessages.GetCount())
			}
			return msgs, nextID, nil
		}
		if waitForMessages {
			clog.waitingReaders.Wait()
		} else {
			if clog.commitIndex == 0 {
				return nil, 0, nil
			}
			return msgs, ID, nil
		}
	}
	return msgs, ID, nil
}
func (clog *CommitLog) Shutdown() {
	clog.lock.RLock()
	defer clog.lock.RUnlock()
	clog.inShutdown = true
	clog.waitingReaders.Broadcast()
}
func (clog *CommitLog) Retrieve(ID int64) (previousIndex int64, previousTerm int64, msgs model.Messages, err error) {
	retrieveID := ID
	if ID >= 2 {
		previousIndex = ID - 1
		previousMessageList, err := clog.log.GetMessages(previousIndex, 1)
		if err != nil {
			return previousIndex, previousTerm, msgs, err
		}
		if previousMessageList.GetCount() > 0 {
			previousTerm, err = previousMessageList.GetMessageTerm(0)
			if err != nil {
				return previousIndex, previousTerm, msgs, err
			}
		} else {
			retrieveID, err = clog.log.GetFirstIndex()
			if err != nil {
				return previousIndex, previousTerm, msgs, err
			}
		}
	} else {
		retrieveID, err = clog.log.GetFirstIndex()
		if err != nil {
			return previousIndex, previousTerm, msgs, err
		}
	}
	msgs, err = clog.log.GetMessages(retrieveID, MAX_RETRIEVE_MESSAGE_COUNT)
	if err != nil {
		clog.node_log("Error reading messages from log in retrieve: %v\n", err)
		return previousIndex, previousTerm, msgs, err
	}
	return previousIndex, previousTerm, msgs, err
}