refactoring
This commit is contained in:
@@ -1,7 +1,6 @@
|
||||
package filer2
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -79,51 +78,3 @@ func (f *Filer) logFlushFunc(startTime, stopTime time.Time, buf []byte) {
|
||||
glog.V(0).Infof("log write failed %s: %v", targetFile, err)
|
||||
}
|
||||
}
|
||||
|
||||
func (f *Filer) ReadLogBuffer(lastReadTime time.Time, eachEventFn func(fullpath string, eventNotification *filer_pb.EventNotification) error) (newLastReadTime time.Time, err error) {
|
||||
|
||||
var bytesBuf *bytes.Buffer
|
||||
bytesBuf = f.MetaLogBuffer.ReadFromBuffer(lastReadTime)
|
||||
if bytesBuf == nil {
|
||||
return
|
||||
}
|
||||
defer f.MetaLogBuffer.ReleaseMeory(bytesBuf)
|
||||
buf := bytesBuf.Bytes()
|
||||
var processedTs int64
|
||||
|
||||
for pos := 0; pos+4 < len(buf); {
|
||||
|
||||
size := util.BytesToUint32(buf[pos : pos+4])
|
||||
entryData := buf[pos+4 : pos+4+int(size)]
|
||||
|
||||
logEntry := &filer_pb.LogEntry{}
|
||||
err = proto.Unmarshal(entryData, logEntry)
|
||||
if err != nil {
|
||||
glog.Errorf("unexpected unmarshal filer_pb.LogEntry: %v", err)
|
||||
return lastReadTime, fmt.Errorf("unexpected unmarshal filer_pb.LogEntry: %v", err)
|
||||
}
|
||||
|
||||
event := &filer_pb.SubscribeMetadataResponse{}
|
||||
err = proto.Unmarshal(logEntry.Data, event)
|
||||
if err != nil {
|
||||
glog.Errorf("unexpected unmarshal filer_pb.SubscribeMetadataResponse: %v", err)
|
||||
return lastReadTime, fmt.Errorf("unexpected unmarshal filer_pb.SubscribeMetadataResponse: %v", err)
|
||||
}
|
||||
|
||||
err = eachEventFn(event.Directory, event.EventNotification)
|
||||
|
||||
processedTs = logEntry.TsNs
|
||||
|
||||
if err != nil {
|
||||
newLastReadTime = time.Unix(0, processedTs)
|
||||
return
|
||||
}
|
||||
|
||||
pos += 4 + int(size)
|
||||
|
||||
}
|
||||
|
||||
newLastReadTime = time.Unix(0, processedTs)
|
||||
return
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user