fix: allow returning negative value from source pending api (#81)

Signed-off-by: Keran Yang <[email protected]>

.github/workflows/ci.yaml

@@ -0,0 +1,27 @@
+name: Go
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+  build:
+    name: Unit Tests
+    runs-on: ubuntu-latest
+    steps:
+
+    - name: Set up Go 1.x
+      uses: actions/setup-go@v2
+      with:
+        go-version: ^1.17
+
+    - name: Check out code into the Go module directory
+      uses: actions/checkout@v2
+
+    - name: Get dependencies
+      run: go mod download
+
+    - name: Test
+      run: make all

pkg/apis/proto/source/v1/source.proto

@@ -113,7 +113,8 @@ message ReadyResponse {
 message PendingResponse {
   message Result {
     // Required field holding the number of pending records at the user defined source.
-    uint64 count = 1;
+    // A negative count indicates that the pending information is not available.
+    int64 count = 1;
   }
   // Required field holding the result.
   Result result = 1;

pkg/sourcer/examples/simple_source/impl/simple_source.go

@@ -25,8 +25,8 @@ func NewSimpleSource() *SimpleSource {
 }
 
 // Pending returns the number of pending records.
-func (s *SimpleSource) Pending(_ context.Context) uint64 {
-	// The simple source always returns 0 to indicate no pending records.
+func (s *SimpleSource) Pending(_ context.Context) int64 {
+	// The simple source always returns zero to indicate no pending records.
 	return 0
 }
 

pkg/sourcer/interface.go

@@ -5,15 +5,16 @@ import (
 	"time"
 )
 
-// Sourcer is the interface for implementation of source.
+// Sourcer is the interface for implementation of the source.
 type Sourcer interface {
 	// Read reads the data from the source and sends the data to the message channel.
 	// Read should never attempt to close the message channel as the caller owns the channel.
 	Read(ctx context.Context, readRequest ReadRequest, messageCh chan<- Message)
 	// Ack acknowledges the data from the source.
 	Ack(ctx context.Context, request AckRequest)
 	// Pending returns the number of pending messages.
-	Pending(ctx context.Context) uint64
+	// When the return value is negative, it indicates the pending information is not available.
+	Pending(ctx context.Context) int64
 }
 
 // ReadRequest is the interface of read request.

pkg/sourcer/options.go

@@ -28,7 +28,7 @@ func WithMaxMessageSize(size int) Option {
 	}
 }
 
-// WithSockAddr start the server with the given sock addr. This is mainly used for testing purpose.
+// WithSockAddr start the server with the given sock addr. This is mainly used for testing purposes.
 func WithSockAddr(addr string) Option {
 	return func(opts *options) {
 		opts.sockAddr = addr

pkg/sourcer/server_test.go

@@ -19,7 +19,7 @@ func (ts TestNoopSource) Ack(ctx context.Context, request AckRequest) {
 	return
 }
 
-func (ts TestNoopSource) Pending(ctx context.Context) uint64 {
+func (ts TestNoopSource) Pending(ctx context.Context) int64 {
 	return 0
 }
 

pkg/sourcer/service_test.go

@@ -18,7 +18,7 @@ import (
 
 var TestEventTime = time.Date(2021, 1, 1, 0, 0, 0, 0, time.UTC)
 var TestKey = "test-key"
-var TestPendingNumber uint64 = 123
+var TestPendingNumber int64 = 123
 
 type TestSource struct{}
 
@@ -32,7 +32,7 @@ func (ts TestSource) Ack(_ context.Context, _ AckRequest) {
 	return
 }
 
-func (ts TestSource) Pending(_ context.Context) uint64 {
+func (ts TestSource) Pending(_ context.Context) int64 {
 	return TestPendingNumber
 }