func (dlog *DiskLogStorage) AppendFirstMessages(msgs model.Messages, leaderFirstIndex int64) (lastIndex int64, err error) {
	dlog.lock.Lock()
	defer dlog.lock.Unlock()
	if msgs.GetCount() == 0 {
		return 0, APPEND_WITH_NO_CONTENT
	}
	if len(dlog.segments) != 1 {
		return 0, FIRST_APPEND_MULTIPLE_SEGMENTS
	}
	seg := dlog.segments[0]
	if seg.getMessageCount() != 0 {
		return 0, errors.New("Attempt to add first messages, but segment already has content!")
	}
	if seg.GetFirstIndex() != leaderFirstIndex {
		dlog.node_log("First index available from leader is %v, re-creating first segment.\n", leaderFirstIndex)
		err = seg.Delete()
		if err != nil {
			dlog.node_log("Error deleting existing segment: %v", err)
		}
		seg, err = CreateNewSegment(dlog.topicName, dlog.pathName, leaderFirstIndex, dlog.target_max_segment_size)
		if err != nil {
			dlog.node_log("Error trying to create a new disk log storage at location %v: %v\n", dlog.pathName, err)
			return 0, err
		}
		dlog.segments[0] = seg
	}
	localAppendIndex, localLastIndex, _, err := seg.AppendMessages(msgs)
	if err != nil {
		dlog.node_log("Error in first append: %v\n", err)
		return 0, err
	}
	dlog.cache.AppendMessages(localAppendIndex, msgs)
	return localLastIndex, nil
}
/*
AppendFirstMessages writes the given messages into a segment and presents it to the cache.

The following steps are followed:

1 - Confirm that only one segment is open
2 - Check whether the firstIndex of the segment matches leaderFirstIndex
3 - If required delete the current segment and create a new one
4 - Append the messages and populate the cache
package disklog
import (
	"errors"
	"github.com/owlfish/forestbus-server/model"
	"github.com/owlfish/forestbus-server/utils"
	"log"
	"path"
	"path/filepath"
	"sort"
	"sync"
	"time"
)
const LOG_NAME_FORMAT = "*-forest-log"
const LOG_NAME_FORMAT_SUFFIX = "-forest-log"
const TARGET_OPEN_SEGMENTS = 2
const SEGMENT_LAST_USED_TIMEOUT = 2 * time.Minute
const SEGMENT_CLEANUP_SCAN_INTERVAL = 1 * time.Hour
const DEFAULT_TARGET_MAX_SEGMENT_SIZE = 1024
const DEFAULT_CACHE_SLOT_SIZE = 30
var APPEND_WITH_NO_CONTENT = errors.New("No messages were given, unable to append.")
var ERR_INDEX_TOO_HIGH = errors.New("Index for message too high")
var FIRST_APPEND_MULTIPLE_SEGMENTS = errors.New("First append attempted, but multiple segments already exist.")
var ERR_SEGMENT_MESSAGE_BODY_UNREADABLE = errors.New("Error reading message body - length in header and available bytes in stream did not match.")
var ERR_SEGMENT_ALREADY_OPEN = errors.New("Segment is already open")
type DiskLogStorage struct {
	topicName                     string
	pathName                      string
	segments                      []*Segment
	cache                         *MessageCache
	lock                          sync.RWMutex
	target_max_segment_size       int
	cache_slot_size               int
	segment_cleanup_age           time.Duration
	closeSegementsChannel         chan int
	segmentCleanupShutdownChannel chan *utils.ShutdownNotifier
	closeSegmentsShutdownChannel  chan *utils.ShutdownNotifier
	node_log                      utils.PrintFFunc
}
type DiskLogStorageInfo struct {
	PathName                string
	SegmentCount            int
	LastIndex               int64
	FirstIndex              int64
	Target_Max_Segment_Size int
	Segment_Cleanup_Age     string
	Cache_Slot_Size         int
	SegmentInfo             []interface{}
	CacheInfo               interface{}
}
type DiskLogConfigFunction func(*DiskLogStorage)
func SetTargetSegmentSize(size int) DiskLogConfigFunction {
	return func(dlog *DiskLogStorage) {
		dlog.target_max_segment_size = size
		if len(dlog.segments) > 0 {
			lastSegment := dlog.segments[len(dlog.segments)-1]
			if lastSegment.GetOpenStatus() {
				lastSegment.target_max_segment_size = dlog.target_max_segment_size
			}
		}
	}
}
func SetCacheSlotSize(size int) DiskLogConfigFunction {
	return func(dlog *DiskLogStorage) {
		dlog.cache_slot_size = size
	}
}
func SetSegmentCleanupAge(age time.Duration) DiskLogConfigFunction {
	return func(dlog *DiskLogStorage) {
		dlog.segment_cleanup_age = age
	}
}
func LoadLog(topicName string, loadpath string, configFuncs ...DiskLogConfigFunction) (*DiskLogStorage, error) {
	dlog := &DiskLogStorage{pathName: loadpath, target_max_segment_size: DEFAULT_TARGET_MAX_SEGMENT_SIZE, cache_slot_size: DEFAULT_CACHE_SLOT_SIZE}
	for _, conf := range configFuncs {
		conf(dlog)
	}
	dlog.node_log = utils.GetTopicLogger(topicName, "DiskLogStorage")
	dlog.topicName = topicName
	dlog.segments = make([]*Segment, 0)
	dlog.cache = CreateCache(dlog.cache_slot_size)
	matches, err := filepath.Glob(path.Join(loadpath, LOG_NAME_FORMAT))
	dlog.node_log("Found %v segment filename matches\n", len(matches))
	if err != nil {
		dlog.node_log("Errror opening disk log: %v\n", err)
		return nil, err
	}
	if len(matches) == 0 {
		dlog.node_log("Creating new disk log in %v\n", loadpath)
		firstSegment, err := CreateNewSegment(dlog.topicName, loadpath, 1, dlog.target_max_segment_size)
		if err != nil {
			dlog.node_log("Error trying to create a new disk log storage at location %v: %v\n", loadpath, err)
			return nil, err
		}
		dlog.segments = append(dlog.segments, firstSegment)
	} else {
		dlog.node_log("Loading existing disk log from %v\n", loadpath)
		sort.Strings(matches)
		for _, filename := range matches {
			dlog.segments = append(dlog.segments, ExistingSegment(topicName, filename, dlog.target_max_segment_size))
		}
		err = dlog.segments[len(dlog.segments)-1].Open(true)
		if err != nil {
			dlog.node_log("Error opening latest segment.")
			return nil, err
		}
	}
	dlog.closeSegementsChannel = make(chan int)
	dlog.closeSegmentsShutdownChannel = make(chan *utils.ShutdownNotifier, 1)
	go dlog.closeSegmentsLoop()
	dlog.segmentCleanupShutdownChannel = make(chan *utils.ShutdownNotifier, 1)
	go dlog.cleanupSegmentsLoop()
	return dlog, err
}
func (dlog *DiskLogStorage) ChangeConfiguration(configFuncs ...DiskLogConfigFunction) {
	dlog.lock.Lock()
	for _, conf := range configFuncs {
		conf(dlog)
	}
	dlog.lock.Unlock()
}
func (dlog *DiskLogStorage) GetMessages(index int64, count int64) (model.Messages, error) {
	if count < 1 {
		return model.EMPTY_MESSAGES, nil
	}
	dlog.lock.RLock()
	defer dlog.lock.RUnlock()
	if dlog.segments[len(dlog.segments)-1].GetLastIndex() < index {
		return model.EMPTY_MESSAGES, nil
	}
	cacheResults := dlog.cache.GetMessages(index, count)
	if cacheResults.GetCount() > 0 {
		return cacheResults, nil
	}
	for i := len(dlog.segments) - 1; i >= 0; i-- {
		seg := dlog.segments[i]
		if seg.GetFirstIndex() <= index {
			msgs, segmentRequiredOpening, err := seg.GetMessages(index, int(count))
			if segmentRequiredOpening {
				dlog.node_log("Segment required opening - kicking off scan for any old segments that can be closed.\n")
				select {
				case dlog.closeSegementsChannel <- 1:
					dlog.node_log("Sent message to close segment loop to scan for segments to close.\n")
				default:
					dlog.node_log("Unable to send message to close segment loop as channel is full\n")
				}
			}
			return msgs, err
		}
	}
	return model.EMPTY_MESSAGES, nil
}
func (dlog *DiskLogStorage) AppendMessages(msgs model.Messages) (lastIndex int64, err error) {
	dlog.lock.Lock()
	defer dlog.lock.Unlock()
	if msgs.GetCount() == 0 {
		return 0, APPEND_WITH_NO_CONTENT
	}
	seg := dlog.segments[len(dlog.segments)-1]
	localAppendIndex, localLastIndex, full, err := seg.AppendMessages(msgs)
	if err != nil {
		return 0, err
	}
	if full {
		dlog.node_log("Creating a new segment with start index %v\n", localLastIndex+1)
		newSegment, err := CreateNewSegment(dlog.topicName, dlog.pathName, localLastIndex+1, dlog.target_max_segment_size)
		if err != nil {
			dlog.node_log("Error trying to create a new disk log storage at location %v: %v\n", dlog.pathName, err)
			return 0, err
		}
		dlog.segments = append(dlog.segments, newSegment)
		dlog.node_log("Re-attempting the append after creating segment.\n")
		localAppendIndex, localLastIndex, full, err = newSegment.AppendMessages(msgs)
		if err != nil {
			dlog.node_log("Error in append: %v\n", err)
			return 0, err
		} else {
			dlog.node_log("Re-attempt at append worked.  Checking whether to close segments.\n")
		}
		select {
		case dlog.closeSegementsChannel <- 1:
			dlog.node_log("Sent message to close segment loop to scan for segments to close.\n")
		default:
			dlog.node_log("Unable to send message to close segment loop as channel is full\n")
		}
	}
	dlog.cache.AppendMessages(localAppendIndex, msgs)
	return localLastIndex, nil
}