func (dlog *DiskLogStorage) ExpVar() interface{} {
	dlog.lock.RLock()
	defer dlog.lock.RUnlock()
	stats := &DiskLogStorageInfo{}
	stats.PathName = dlog.pathName
	stats.SegmentCount = len(dlog.segments)
	stats.SegmentInfo = make([]interface{}, stats.SegmentCount)
	for i, seg := range dlog.segments {
		stats.SegmentInfo[i] = seg.ExpVar()
	}
	stats.CacheInfo = dlog.cache.ExpVar()
	stats.LastIndex = dlog.segments[len(dlog.segments)-1].GetLastIndex()
	stats.FirstIndex = dlog.segments[0].GetFirstIndex()
	stats.Target_Max_Segment_Size = dlog.target_max_segment_size
	stats.Cache_Slot_Size = dlog.cache_slot_size
	stats.Segment_Cleanup_Age = dlog.segment_cleanup_age.String()
	return stats
}
Returns stats for expvar
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
}
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
}
func (dlog *DiskLogStorage) Sync() error {
	dlog.lock.Lock()
	defer dlog.lock.Unlock()
	err := dlog.segments[len(dlog.segments)-1].Sync()
	if err != nil {
		return err
	}
	if len(dlog.segments) >= 2 {
		err := dlog.segments[len(dlog.segments)-2].Sync()
		if err != nil {
			return err
		}
	}
	return nil
}
func (dlog *DiskLogStorage) GetLastIndex() (lastIndex int64, err error) {
	dlog.lock.RLock()
	defer dlog.lock.RUnlock()
	lastIndex = dlog.segments[len(dlog.segments)-1].GetLastIndex()
	return lastIndex, nil
}
func (dlog *DiskLogStorage) GetFirstIndex() (firstIndex int64, err error) {
	dlog.lock.RLock()
	defer dlog.lock.RUnlock()
	firstIndex = dlog.segments[0].GetFirstIndex()
	return firstIndex, nil
}
func (dlog *DiskLogStorage) TruncateMessages(index int64) error {
	if index < 1 {
		return nil
	}
	dlog.lock.Lock()
	defer dlog.lock.Unlock()
	dlog.node_log("TruncateMessages called for index %v\n", index)
	dlog.cache.Truncate(index)
	lastSegmentToKeep := len(dlog.segments) - 1
	for i := 0; i < len(dlog.segments); i++ {
		if dlog.segments[i].GetLastIndex() >= index {
			if dlog.segments[i].GetFirstIndex() >= index {
				if i == 0 {
					dlog.node_log("Found first segment should be deleted, truncating instead\n")
					dlog.segments[i].Truncate(index)
				} else {
					if lastSegmentToKeep >= i {
						lastSegmentToKeep = i - 1
						dlog.node_log("Found segments beyond %v (%v) can be deleted\n", lastSegmentToKeep, dlog.segments[i].filename)
					}
				}
			} else {
				dlog.node_log("Found only part of segment (%v) needs to be truncated.\n", dlog.segments[i].filename)
				err := dlog.segments[i].Truncate(index)
				if err != nil {
					dlog.node_log("Error during truncate of segment (%v): %v\n", dlog.segments[i].filename, err)
					return err
				}
			}
		}
	}
	for i := lastSegmentToKeep + 1; i < len(dlog.segments); i++ {
		if i > 0 {
			dlog.node_log("Truncate requires removing a segment %v\n", dlog.segments[i].filename)
			err := dlog.segments[i].Delete()
			if err != nil {
				dlog.node_log("Error removing segment file.\n")
				return err
			}
		}
	}
	if lastSegmentToKeep > 0 {
		dlog.segments = dlog.segments[:lastSegmentToKeep+1]
	}
	return nil
}