Topics progress - receive and commit

Author: Tim Middleton

Makefile

@@ -37,7 +37,7 @@ MVN_VERSION ?= 1.0.0
 SHELL := /bin/bash
 
 # Coherence CE version to run base tests against
-COHERENCE_VERSION ?= 22.06.12
+COHERENCE_VERSION ?= 22.06.13
 COHERENCE_GROUP_ID ?= com.oracle.coherence.ce
 COHERENCE_WKA1 ?= server1
 COHERENCE_WKA2 ?= server1
@@ -225,8 +225,8 @@ ifeq ($(SKIP_PROTO_GENERATION),true)
 	@echo "Skipping proto generation..."
 else
 	mkdir -p $(PROTO_DIR) || true
-	curl $(CURL_AUTH) -o $(PROTO_DIR)/services.proto https://raw.githubusercontent.com/oracle/coherence/22.06.12/prj/coherence-grpc/src/main/proto/services.proto
-	curl $(CURL_AUTH) -o $(PROTO_DIR)/messages.proto https://raw.githubusercontent.com/oracle/coherence/22.06.12/prj/coherence-grpc/src/main/proto/messages.proto
+	curl $(CURL_AUTH) -o $(PROTO_DIR)/services.proto https://raw.githubusercontent.com/oracle/coherence/22.06.13/prj/coherence-grpc/src/main/proto/services.proto
+	curl $(CURL_AUTH) -o $(PROTO_DIR)/messages.proto https://raw.githubusercontent.com/oracle/coherence/22.06.13/prj/coherence-grpc/src/main/proto/messages.proto
 	echo "" >> $(PROTO_DIR)/services.proto
 	echo "" >> $(PROTO_DIR)/messages.proto
 	echo 'option go_package = "github.com/oracle/coherence-go-client/proto";' >> $(PROTO_DIR)/services.proto

coherence/subscriber/subscriber.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"github.com/oracle/coherence-go-client/v2/coherence/filters"
 	pb1topics "github.com/oracle/coherence-go-client/v2/proto/topics"
+	"time"
 )
 
 type ReceiveStatus int32
@@ -29,15 +30,28 @@ type EnsureSubscriberResult struct {
 	UUID              string
 }
 
-type ReceiveResult[V any] struct {
-	Status ReceiveStatus
-}
-
 // Options provides options for creating a subscriber.
 type Options struct {
 	SubscriberGroup *string
 	Filter          filters.Filter
 	Extractor       []byte
+	AutoCommit      bool
+	MaxMessages     int32
+}
+
+// CommitResponse represents th response from a [Subscriber] commit.
+type CommitResponse struct {
+	Channel  int32
+	Position *pb1topics.TopicPosition
+	Head     *pb1topics.TopicPosition
+}
+
+// ReceiveResponse represents a response from a [Subscriber] receive request.
+type ReceiveResponse[V any] struct {
+	Channel   int32
+	Value     *V
+	Position  *pb1topics.TopicPosition
+	Timestamp time.Time
 }
 
 // TODO: Additional options
@@ -61,6 +75,20 @@ func WithFilter(fltr filters.Filter) func(options *Options) {
 	}
 }
 
+// WithAutoCommit returns a function to set auto commit to be true for a [Subscriber].
+func WithAutoCommit() func(options *Options) {
+	return func(s *Options) {
+		s.AutoCommit = true
+	}
+}
+
+// WithMaxMessages returns a function to set the maximum messages for a[Subscriber].
+func WithMaxMessages(maxMessages int32) func(options *Options) {
+	return func(s *Options) {
+		s.MaxMessages = maxMessages
+	}
+}
+
 // WithTransformer returns a function to set the extractor [Subscriber].
 func WithTransformer(extractor []byte) func(options *Options) {
 	return func(s *Options) {