sdjournal: add support for cursors (get&seek&test)

Author: glerchundi

sdjournal/journal.go

@@ -155,6 +155,24 @@ package sdjournal
 // }
 //
 // int
+// my_sd_journal_get_cursor(void *f, sd_journal *j, char **cursor)
+// {
+//   int (*sd_journal_get_cursor)(sd_journal *, char **);
+//
+//   sd_journal_get_cursor = f;
+//   return sd_journal_get_cursor(j, cursor);
+// }
+//
+// int
+// my_sd_journal_test_cursor(void *f, sd_journal *j, const char *cursor)
+// {
+//   int (*sd_journal_test_cursor)(sd_journal *, const char *);
+//
+//   sd_journal_test_cursor = f;
+//   return sd_journal_test_cursor(j, cursor);
+// }
+//
+// int
 // my_sd_journal_get_realtime_usec(void *f, sd_journal *j, uint64_t *usec)
 // {
 //   int (*sd_journal_get_realtime_usec)(sd_journal *, uint64_t *);
@@ -181,6 +199,16 @@ package sdjournal
 //   return sd_journal_seek_tail(j);
 // }
 //
+//
+// int
+// my_sd_journal_seek_cursor(void *f, sd_journal *j, const char *cursor)
+// {
+//   int (*sd_journal_seek_cursor)(sd_journal *, const char *);
+//
+//   sd_journal_seek_cursor = f;
+//   return sd_journal_seek_cursor(j, cursor);
+// }
+//
 // int
 // my_sd_journal_seek_realtime_usec(void *f, sd_journal *j, uint64_t usec)
 // {
@@ -596,6 +624,50 @@ func (j *Journal) GetRealtimeUsec() (uint64, error) {
 	return uint64(usec), nil
 }
 
+// GetCursor gets the cursor of the current journal entry.
+func (j *Journal) GetCursor() (string, error) {
+	var d *C.char
+
+	sd_journal_get_cursor, err := j.getFunction("sd_journal_get_cursor")
+	if err != nil {
+		return "", err
+	}
+
+	j.mu.Lock()
+	r := C.my_sd_journal_get_cursor(sd_journal_get_cursor, j.cjournal, &d)
+	j.mu.Unlock()
+
+	if r < 0 {
+		return "", fmt.Errorf("failed to get cursor: %d", syscall.Errno(-r))
+	}
+
+	cursor := C.GoString(d)
+
+	return cursor, nil
+}
+
+// TestCursor checks whether the current position in the journal matches the
+// specified cursor
+func (j *Journal) TestCursor(cursor string) error {
+	sd_journal_test_cursor, err := j.getFunction("sd_journal_test_cursor")
+	if err != nil {
+		return err
+	}
+
+	c := C.CString(cursor)
+	defer C.free(unsafe.Pointer(c))
+
+	j.mu.Lock()
+	r := C.my_sd_journal_test_cursor(sd_journal_test_cursor, j.cjournal, c)
+	j.mu.Unlock()
+
+	if r < 0 {
+		return fmt.Errorf("failed to test to cursor %q: %d", cursor, syscall.Errno(-r))
+	}
+
+	return nil
+}
+
 // SeekHead seeks to the beginning of the journal, i.e. the oldest available
 // entry.
 func (j *Journal) SeekHead() error {
@@ -653,6 +725,27 @@ func (j *Journal) SeekRealtimeUsec(usec uint64) error {
 	return nil
 }
 
+// SeekCursor seeks to a concrete journal cursor.
+func (j *Journal) SeekCursor(cursor string) error {
+	sd_journal_seek_cursor, err := j.getFunction("sd_journal_seek_cursor")
+	if err != nil {
+		return err
+	}
+
+	c := C.CString(cursor)
+	defer C.free(unsafe.Pointer(c))
+
+	j.mu.Lock()
+	r := C.my_sd_journal_seek_cursor(sd_journal_seek_cursor, j.cjournal, c)
+	j.mu.Unlock()
+
+	if r < 0 {
+		return fmt.Errorf("failed to seek to cursor %q: %d", cursor, syscall.Errno(-r))
+	}
+
+	return nil
+}
+
 // Wait will synchronously wait until the journal gets changed. The maximum time
 // this call sleeps may be controlled with the timeout parameter.  If
 // sdjournal.IndefiniteWait is passed as the timeout parameter, Wait will

sdjournal/journal_test.go

@@ -16,6 +16,9 @@
 package sdjournal
 
 import (
+	"errors"
+	"fmt"
+	"io"
 	"os"
 	"testing"
 	"time"
@@ -88,3 +91,62 @@ func TestJournalGetUsage(t *testing.T) {
 		t.Fatalf("Error getting journal size: %s", err)
 	}
 }
+
+func TestJournalCursorGetSeekAndTest(t *testing.T) {
+	j, err := NewJournal()
+	if err != nil {
+		t.Fatalf("Error opening journal: %s", err)
+	}
+
+	if j == nil {
+		t.Fatal("Got a nil journal")
+	}
+
+	defer j.Close()
+
+	waitAndNext := func(j *Journal) error {
+		r := j.Wait(time.Duration(1) * time.Second)
+		if r < 0 {
+			return errors.New("Error waiting to journal")
+		}
+
+		n, err := j.Next()
+		if err != nil {
+			return fmt.Errorf("Error reading to journal: %s", err)
+		}
+
+		if n == 0 {
+			return fmt.Errorf("Error reading to journal: %s", io.EOF)
+		}
+
+		return nil
+	}
+
+	err = journal.Print(journal.PriInfo, "test message for cursor %s", time.Now())
+	if err != nil {
+		t.Fatalf("Error writing to journal: %s", err)
+	}
+
+	if err = waitAndNext(j); err != nil {
+		t.Fatalf(err.Error())
+	}
+
+	c, err := j.GetCursor()
+	if err != nil {
+		t.Fatalf("Error getting cursor from journal: %s", err)
+	}
+
+	err = j.SeekCursor(c)
+	if err != nil {
+		t.Fatalf("Error seeking cursor to journal: %s", err)
+	}
+
+	if err = waitAndNext(j); err != nil {
+		t.Fatalf(err.Error())
+	}
+
+	err = j.TestCursor(c)
+	if err != nil {
+		t.Fatalf("Error testing cursor to journal: %s", err)
+	}
+}

sdjournal/read.go

@@ -33,6 +33,7 @@ type JournalReaderConfig struct {
 	// where the reading begins within the journal.
 	Since       time.Duration // start relative to a Duration from now
 	NumFromTail uint64        // start relative to the tail
+	Cursor      string        // start relative to the cursor
 
 	// Show only journal entries whose fields match the supplied values. If
 	// the array is empty, entries will not be filtered.
@@ -89,6 +90,11 @@ func NewJournalReader(config JournalReaderConfig) (*JournalReader, error) {
 		if _, err := r.journal.PreviousSkip(config.NumFromTail + 1); err != nil {
 			return nil, err
 		}
+	} else if config.Cursor != "" {
+		// Start based on a custom cursor
+		if err := r.journal.SeekCursor(config.Cursor); err != nil {
+			return nil, err
+		}
 	}
 
 	return r, nil