Merge pull request #53 from stealthrocket/record-offset

internal/timemachine/log_test.go

@@ -70,6 +70,14 @@ func TestReadRecordBatch(t *testing.T) {
  firstOffset += int64(len(batch))
  }
 
+ recordOffset := int64(0)
+ for _, batch := range batches {
+ for i := range batch {
+ batch[i].Offset = recordOffset
+ recordOffset++
+ }
+ }
+
  reader := timemachine.NewLogReader(bytes.NewReader(buffer.Bytes()), startTime)
  batchesRead := make([][]timemachine.Record, 0, len(batches))
  for {

internal/timemachine/record.go

@@ -11,6 +11,7 @@ import (
 
 // Record is a read-only record from the log.
 type Record struct {
+ Offset int64
  Time time.Time
  FunctionID int
  FunctionCall []byte

internal/timemachine/record_batch.go

@@ -32,6 +32,7 @@ type RecordBatch struct {
  // When reading records from the batch, this holds the current offset into
  // the records buffer.
  offset uint32
+ index uint32
 }
 
 // Reset resets the record batch.
@@ -55,6 +56,7 @@ func (b *RecordBatch) Reset(startTime time.Time, buf []byte, reader io.Reader) {
  b.reader.N = 0
  }
  b.offset = 0
+ b.index = 0
 }
 
 // Size is the size of the adjacent record data.
@@ -133,11 +135,13 @@ func (b *RecordBatch) Read(records []Record) (int, error) {
  j := 4 + b.offset + size
  r := logsegment.GetRootAsRecord(batch[i:j:j], 0)
  records[n] = Record{
+ Offset: b.FirstOffset() + int64(b.index),
  Time: b.startTime.Add(time.Duration(r.Timestamp())),
  FunctionID: int(r.FunctionId()),
  FunctionCall: r.FunctionCallBytes(),
  }
  b.offset += size + 4
+ b.index++
  }
  return len(records), nil
 }