Unified scheduler

Author: dim

.golangci.yml

@@ -6,6 +6,8 @@ linters:
         - (*github.com/bsm/feedx.Writer).Discard
         - (*github.com/bsm/feedx.Reader).Close
         - (github.com/bsm/feedx.Consumer).Close
+        - (*github.com/bsm/feedx.IncrementalProducer).Close
         - (*github.com/bsm/feedx.Producer).Close
+        - (*github.com/bsm/bfs.InMem).Close
         - (*github.com/bsm/bfs.Object).Close
         - (github.com/bsm/bfs.Writer).Discard

Gemfile.lock

@@ -49,7 +49,7 @@ GEM
       rubocop-ast (>= 1.47.1, < 2.0)
       ruby-progressbar (~> 1.7)
       unicode-display_width (>= 2.4.0, < 4.0)
-    rubocop-ast (1.47.1)
+    rubocop-ast (1.48.0)
       parser (>= 3.3.7.2)
       prism (~> 1.4)
     rubocop-rake (0.7.1)

consumer.go

@@ -8,18 +8,6 @@ import (
 	"github.com/bsm/bfs"
 )
 
-// ConsumeStatus is returned by Consumer instances.
-type ConsumeStatus struct {
-	// Skipped indicates the the sync was skipped, because there were no new changes.
-	Skipped bool
-	// Version indicates the current version of the remote feed.
-	Version int64
-	// PreviousVersion indicates the last known version to the consumer before the sync.
-	PreviousVersion int64
-	// NumRead returns the number of items read.
-	NumRead int64
-}
-
 // ConsumeFunc is a callback invoked by consumers.
 type ConsumeFunc func(context.Context, *Reader) error
 
@@ -28,7 +16,7 @@ type ConsumeFunc func(context.Context, *Reader) error
 type Consumer interface {
 	// Consume initiates a sync attempt. It will consume the remote feed only if it has changed since
 	// last invocation.
-	Consume(context.Context, *ReaderOptions, ConsumeFunc) (*ConsumeStatus, error)
+	Consume(context.Context, *ReaderOptions, ConsumeFunc) (*Status, error)
 
 	// Version indicates the most recently consumed version.
 	Version() int64
@@ -86,21 +74,21 @@ type consumer struct {
 }
 
 // Consume implements Consumer interface.
-func (c *consumer) Consume(ctx context.Context, opt *ReaderOptions, fn ConsumeFunc) (*ConsumeStatus, error) {
-	prevVersion := c.Version()
-	status := ConsumeStatus{
-		PreviousVersion: prevVersion,
+func (c *consumer) Consume(ctx context.Context, opt *ReaderOptions, fn ConsumeFunc) (*Status, error) {
+	localVersion := c.Version()
+	status := Status{
+		LocalVersion: localVersion,
 	}
 
 	// retrieve remote mtime
-	version, err := fetchRemoteVersion(ctx, c.remote)
+	remoteVersion, err := fetchRemoteVersion(ctx, c.remote)
 	if err != nil {
 		return nil, err
 	}
-	status.Version = version
+	status.RemoteVersion = remoteVersion
 
 	// skip sync unless modified
-	if prevVersion > 0 && prevVersion == version {
+	if skipSync(remoteVersion, localVersion) {
 		status.Skipped = true
 		return &status, nil
 	}
@@ -122,8 +110,8 @@ func (c *consumer) Consume(ctx context.Context, opt *ReaderOptions, fn ConsumeFu
 		return nil, err
 	}
 
-	status.NumRead = reader.NumRead()
-	c.version.Store(version)
+	status.NumItems = reader.NumRead()
+	c.version.Store(remoteVersion)
 	return &status, nil
 }
 

consumer_test.go

@@ -12,66 +12,66 @@ import (
 
 func TestConsumer(t *testing.T) {
 	t.Run("consumes", func(t *testing.T) {
-		csm := fixConsumer(t, 33)
+		csm := fixConsumer(t, 101)
 		defer csm.Close()
 
 		if exp, got := int64(0), csm.Version(); exp != got {
 			t.Errorf("expected %v, got %v", exp, got)
 		}
 
 		// first attempt
-		msgs := testConsume(t, csm, &feedx.ConsumeStatus{
-			PreviousVersion: 0,
-			Version:         33,
-			Skipped:         false,
-			NumRead:         2,
+		msgs := testConsume(t, csm, &feedx.Status{
+			LocalVersion:  0,
+			RemoteVersion: 101,
+			Skipped:       false,
+			NumItems:      2,
 		})
-		if exp, got := int64(33), csm.Version(); exp != got {
+		if exp, got := int64(101), csm.Version(); exp != got {
 			t.Errorf("expected %v, got %v", exp, got)
 		}
 		if exp, got := 2, len(msgs); exp != got {
 			t.Errorf("expected %v, got %v", exp, got)
 		}
 
 		// second attempt
-		_ = testConsume(t, csm, &feedx.ConsumeStatus{
-			PreviousVersion: 33,
-			Version:         33,
-			Skipped:         true,
-			NumRead:         0,
+		_ = testConsume(t, csm, &feedx.Status{
+			LocalVersion:  101,
+			RemoteVersion: 101,
+			Skipped:       true,
+			NumItems:      0,
 		})
 	})
 
 	t.Run("always if no version", func(t *testing.T) {
 		csm := fixConsumer(t, 0)
 		defer csm.Close()
 
-		testConsume(t, csm, &feedx.ConsumeStatus{NumRead: 2})
-		testConsume(t, csm, &feedx.ConsumeStatus{NumRead: 2})
+		testConsume(t, csm, &feedx.Status{NumItems: 2})
+		testConsume(t, csm, &feedx.Status{NumItems: 2})
 	})
 
 	t.Run("incremental", func(t *testing.T) {
-		csm := fixIncrementalConsumer(t, 33)
+		csm := fixIncrementalConsumer(t, 101)
 		defer csm.Close()
 
 		// first attempt
-		msgs := testConsume(t, csm, &feedx.ConsumeStatus{
-			PreviousVersion: 0,
-			Version:         33,
-			NumRead:         4,
+		msgs := testConsume(t, csm, &feedx.Status{
+			LocalVersion:  0,
+			RemoteVersion: 101,
+			NumItems:      4,
 		})
-		if exp, got := int64(33), csm.Version(); exp != got {
+		if exp, got := int64(101), csm.Version(); exp != got {
 			t.Errorf("expected %v, got %v", exp, got)
 		}
 		if exp, got := 4, len(msgs); exp != got {
 			t.Errorf("expected %v, got %v", exp, got)
 		}
 
 		// second attempt
-		_ = testConsume(t, csm, &feedx.ConsumeStatus{
-			PreviousVersion: 33,
-			Version:         33,
-			Skipped:         true,
+		_ = testConsume(t, csm, &feedx.Status{
+			LocalVersion:  101,
+			RemoteVersion: 101,
+			Skipped:       true,
 		})
 	})
 
@@ -81,9 +81,12 @@ func fixConsumer(t *testing.T, version int64) feedx.Consumer {
 	t.Helper()
 
 	obj := bfs.NewInMemObject("path/to/file.json")
+	t.Cleanup(func() { _ = obj.Close() })
+
 	if err := writeN(obj, 2, version); err != nil {
 		t.Fatal("unexpected error", err)
 	}
+
 	csm := feedx.NewConsumerForRemote(obj)
 	t.Cleanup(func() { _ = csm.Close() })
 
@@ -128,7 +131,7 @@ func fixIncrementalConsumer(t *testing.T, version int64) feedx.Consumer {
 	return csm
 }
 
-func testConsume(t *testing.T, csm feedx.Consumer, exp *feedx.ConsumeStatus) (msgs []*testdata.MockMessage) {
+func testConsume(t *testing.T, csm feedx.Consumer, exp *feedx.Status) (msgs []*testdata.MockMessage) {
 	t.Helper()
 
 	status, err := csm.Consume(t.Context(), nil, func(ctx context.Context, r *feedx.Reader) (err error) {

example_test.go

@@ -59,7 +59,7 @@ func ExampleConsumer() {
 		panic(err)
 	}
 
-	fmt.Printf("STATUS skipped:%v version:%v read:%v\n", status.Skipped, status.Version, status.NumRead)
+	fmt.Printf("STATUS skipped:%v version:%v read:%v\n", status.Skipped, status.RemoteVersion, status.NumItems)
 	fmt.Printf("DATA   %v\n", todos)
 
 	// Output:
@@ -80,21 +80,21 @@ func ExampleScheduler() {
 
 	job := feedx.Every(time.Hour).
 		WithContext(ctx).
-		BeforeConsume(func() bool {
-			fmt.Println("[H] BeforeConsume")
+		BeforeSync(func() bool {
+			fmt.Println("[H] Before sync")
 			return true
 		}).
-		AfterConsume(func(_ *feedx.ConsumeStatus, err error) {
-			fmt.Printf("[H] AfterConsume - error:%v", err)
+		AfterSync(func(_ *feedx.Status, err error) {
+			fmt.Printf("[H] After sync - error:%v", err)
 		}).
 		Consume(csm, func(_ context.Context, _ *feedx.Reader) error {
 			fmt.Println("[*] Consuming feed")
 			return nil
 		})
-	defer job.Stop()
+	job.Stop()
 
 	// Output:
-	// [H] BeforeConsume
+	// [H] Before sync
 	// [*] Consuming feed
-	// [H] AfterConsume - error:<nil>
+	// [H] After sync - error:<nil>
 }

feedx.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"strconv"
-	"time"
 
 	"github.com/bsm/bfs"
 )
@@ -26,13 +25,18 @@ func fetchRemoteVersion(ctx context.Context, obj *bfs.Object) (int64, error) {
 	return version, nil
 }
 
-func epochToTime(epoch int64) time.Time {
-	return time.Unix(epoch/1000, epoch%1000*1e6)
+// Status is returned by sync processes.
+type Status struct {
+	// Skipped indicates the the sync was skipped, because there were no new changes.
+	Skipped bool
+	// LocalVersion indicates the local version before sync.
+	LocalVersion int64
+	// RemoteVersion indicates the remote version before sync.
+	RemoteVersion int64
+	// NumItems returns the number of items processed, either read of written.
+	NumItems int64
 }
 
-func timeToEpoch(t time.Time) int64 {
-	if n := t.Unix()*1000 + int64(t.Nanosecond()/1e6); n > 0 {
-		return n
-	}
-	return 0
+func skipSync(srcVersion, targetVersion int64) bool {
+	return (srcVersion != 0 || targetVersion != 0) && srcVersion <= targetVersion
 }