Additional changes/ tests

Author: Tim Middleton

Makefile

@@ -449,7 +449,7 @@ test-e2e-standalone-queues: test-clean test gotestsum $(BUILD_PROPS) ## Run e2e
 # ----------------------------------------------------------------------------------------------------------------------
 .PHONY: test-e2e-standalone-topics
 test-e2e-standalone-topics: test-clean test gotestsum $(BUILD_PROPS) ## Run e2e tests with Coherence queues
-	cd test && CGO_ENABLED=0 $(GOTESTSUM) --format testname --junitfile $(TEST_LOGS_DIR)/go-client-test-queues.xml \
+	cd test && CGO_ENABLED=0 $(GOTESTSUM) --format testname --junitfile $(TEST_LOGS_DIR)/go-client-test-topics.xml \
 	  -- $(GO_TEST_FLAGS) -v -coverprofile=$(COVERAGE_DIR)/cover-functional-topics.out -v ./e2e/topics/... -coverpkg=github.com/oracle/coherence-go-client/v2/coherence/...
 	go tool cover -func=$(COVERAGE_DIR)/cover-functional-topics.out | grep -v '0.0%'
 

coherence/session.go

@@ -276,6 +276,8 @@ func NewSession(ctx context.Context, options ...func(session *SessionOptions)) (
 // setLogLevel sets the log level from the COHERENCE_LOG_LEVEL environment variable.
 func setLogLevel(envLevel string) {
 	var level int
+	logLevelMutex.Lock()
+	defer logLevelMutex.Unlock()
 
 	// try to convert from integer first
 	if lvl, err := strconv.Atoi(envLevel); err == nil {

coherence/subscriber/subscriber.go

@@ -37,6 +37,7 @@ type Options struct {
 	Extractor       []byte
 	AutoCommit      bool
 	MaxMessages     int32
+	Channels        []int32
 }
 
 // CommitResponse represents th response from a [Subscriber] commit.
@@ -82,13 +83,20 @@ func WithAutoCommit() func(options *Options) {
 	}
 }
 
-// WithMaxMessages returns a function to set the maximum messages for a[Subscriber].
+// 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
 	}
 }
 
+// WithChannels returns a function to set the channels for a [Subscriber].
+func WithChannels(channels []int32) func(options *Options) {
+	return func(s *Options) {
+		s.Channels = channels
+	}
+}
+
 // WithTransformer returns a function to set the extractor [Subscriber].
 func WithTransformer(extractor []byte) func(options *Options) {
 	return func(s *Options) {
@@ -97,5 +105,10 @@ func WithTransformer(extractor []byte) func(options *Options) {
 }
 
 func (o *Options) String() string {
-	return fmt.Sprintf("options{SubscriberGroup=%v, filter=%v}", o.SubscriberGroup, o.Filter)
+	var subGroup string
+	if o.SubscriberGroup != nil {
+		subGroup = *o.SubscriberGroup
+	}
+	return fmt.Sprintf("options{SubscriberGroup=%v, filter=%v, MaxMessages=%v, AutoCommit=%v, Transformer=%v, Channels=%v}",
+		subGroup, o.Filter, o.MaxMessages, o.AutoCommit, o.Extractor, o.Channels)
 }

coherence/topics.go

@@ -71,7 +71,6 @@ type NamedTopic[V any] interface {
 	CreateSubscriberGroup(ctx context.Context, subscriberGroup string, options ...func(o *subscribergroup.Options)) error
 
 	// DestroySubscriberGroup destroys a subscriber group.
-	// TODO: Not viable to implement?
 	DestroySubscriberGroup(ctx context.Context, subscriberGroup string) error
 
 	// AddLifecycleListener adds a [TopicLifecycleListener] to this topic.
@@ -114,9 +113,13 @@ type Subscriber[V any] interface {
 	// If the length of the [subscriber.ReceiveResponse] is zero this means no messages were available.
 	Receive(ctx context.Context) ([]*subscriber.ReceiveResponse[V], error)
 
-	// Commit commits a channel and position meaning the message will be removed from the topic.
+	// Commit commits a channel and position meaning the message will not be get redelivered if the subscriber gets disconnected.
 	Commit(ctx context.Context, channel int32, position *pb1topics.TopicPosition) (*subscriber.CommitResponse, error)
 
+	// Peek attempts to peek messages without removing them. If the length of the [subscriber.ReceiveResponse]
+	//is zero this means no messages were available.
+	Peek(ctx context.Context) ([]*subscriber.ReceiveResponse[V], error)
+
 	// DestroySubscriberGroup destroys a subscriber group created by this subscriber.
 	DestroySubscriberGroup(ctx context.Context, subscriberGroup string) error
 
@@ -296,7 +299,30 @@ func (bt *baseTopicsClient[V]) CreateSubscriberGroup(ctx context.Context, subscr
 }
 
 func (bt *baseTopicsClient[V]) DestroySubscriberGroup(ctx context.Context, subscriberGroup string) error {
-	return DestroySubscriberGroup(ctx, bt.session, subscriberGroup)
+	return DestroySubscriberGroup(ctx, bt.session, bt.topicID, subscriberGroup)
+}
+
+func (bt *baseTopicsClient[V]) isTopicDestroyed() bool {
+	bt.mutex.RLock()
+	defer bt.mutex.RUnlock()
+	return bt.isDestroyed
+}
+
+func (bt *baseTopicsClient[V]) setDestroyed() {
+	bt.mutex.Lock()
+	defer bt.mutex.Unlock()
+	bt.isDestroyed = true
+}
+func (bt *baseTopicsClient[V]) isTopicReleased() bool {
+	bt.mutex.RLock()
+	defer bt.mutex.RUnlock()
+	return bt.isReleased
+}
+
+func (bt *baseTopicsClient[V]) setReleased() {
+	bt.mutex.Lock()
+	defer bt.mutex.Unlock()
+	bt.isReleased = true
 }
 
 func newPublisher[V any](session *Session, bt *baseTopicsClient[V], result *publisher.EnsurePublisherResult, topicName string, options *publisher.Options) (Publisher[V], error) {
@@ -385,12 +411,12 @@ func closeSubscriber[V any](ctx context.Context, ts *topicSubscriber[V]) error {
 	return ts.session.v1StreamManagerTopics.destroyPublisherOrSubscriber(ctx, ts.proxyID, pb1topics.TopicServiceRequestType_DestroySubscriber)
 }
 
-func receiveInternal[V any](ctx context.Context, ts *topicSubscriber[V], maxMessages *int32) ([]*pb1topics.TopicElement, error) {
+func peekOrReceiveInternal[V any](ctx context.Context, ts *topicSubscriber[V], maxMessages *int32, isReceive bool) ([]*pb1topics.TopicElement, error) {
 	if ts.isClosed {
 		return nil, ErrSubscriberClosed
 	}
 
-	return ts.session.v1StreamManagerTopics.receive(ctx, ts.proxyID, maxMessages)
+	return ts.session.v1StreamManagerTopics.peekOrReceive(ctx, ts.proxyID, maxMessages, isReceive)
 }
 
 func commitInternal[V any](ctx context.Context, ts *topicSubscriber[V], channel int32, position *pb1topics.TopicPosition) (*subscriber.CommitResponse, error) {
@@ -405,16 +431,24 @@ func (ts *topicSubscriber[V]) DestroySubscriberGroup(ctx context.Context, subscr
 	return ts.session.v1StreamManagerTopics.destroySubscriberGroup(ctx, ts.proxyID, subscriberGroup)
 }
 
-// Receive attempts to receive messages. Messages must be committed to be considered processed unless autoCommit is set.
+func (ts *topicSubscriber[V]) Peek(ctx context.Context) ([]*subscriber.ReceiveResponse[V], error) {
+	return ts.peekOrReceive(ctx, false)
+}
+
 func (ts *topicSubscriber[V]) Receive(ctx context.Context) ([]*subscriber.ReceiveResponse[V], error) {
-	r, err := receiveInternal(ctx, ts, &ts.options.MaxMessages)
+	return ts.peekOrReceive(ctx, true)
+}
+
+// Receive attempts to peekOrReceive messages. Messages must be committed to be considered processed unless autoCommit is set.
+func (ts *topicSubscriber[V]) peekOrReceive(ctx context.Context, isReceive bool) ([]*subscriber.ReceiveResponse[V], error) {
+	r, err := peekOrReceiveInternal(ctx, ts, &ts.options.MaxMessages, isReceive)
 	if err != nil {
 		return nil, err
 	}
 
 	values, err := ts.decodeAll(r)
 	if err != nil {
-		return nil, fmt.Errorf("unable to decode receive values: %v", err)
+		return nil, fmt.Errorf("unable to decode peek rr receive values: %v", err)
 	}
 
 	if ts.options.AutoCommit && len(values) > 0 {
@@ -530,7 +564,6 @@ func CreatSubscriberWithTransformer[E any](ctx context.Context, session *Session
 func CreateSubscriber[V any](ctx context.Context, session *Session, topicName string, options ...func(cache *subscriber.Options)) (Subscriber[V], error) {
 	var (
 		subscriberOptions = &subscriber.Options{}
-		binFilter         []byte
 		err               error
 	)
 
@@ -545,18 +578,15 @@ func CreateSubscriber[V any](ctx context.Context, session *Session, topicName st
 		}
 	}
 
-	if subscriberOptions.Filter != nil {
-		binFilter, err = session.genericSerializer.Serialize(subscriberOptions.Filter)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	result, err := session.v1StreamManagerTopics.ensureSubscriber(ctx, topicName, subscriberOptions.SubscriberGroup, binFilter)
+	result, err := session.v1StreamManagerTopics.ensureSubscriber(ctx, topicName, subscriberOptions)
 	if err != nil {
 		return nil, err
 	}
 
+	if result == nil {
+		return nil, fmt.Errorf("unable to create subscriber with options %v, check server logs: %v", subscriberOptions, err)
+	}
+
 	return newSubscriber[V](session, nil, result, topicName, subscriberOptions)
 }
 
@@ -590,15 +620,14 @@ func CreateSubscriberGroup(ctx context.Context, session *Session, topicName stri
 }
 
 // DestroySubscriberGroup destroys a subscriber group.
-func DestroySubscriberGroup(ctx context.Context, session *Session, subscriberGroup string) error {
+func DestroySubscriberGroup(ctx context.Context, session *Session, proxyID int32, subscriberGroup string) error {
 	if session.v1StreamManagerTopics == nil {
 		if err := ensureV1StreamManagerTopics(session); err != nil {
 			return err
 		}
 	}
 
-	// TODO: Is this valid???, how can we determine the proxyID if we don't have a subscriber
-	return session.v1StreamManagerTopics.destroySubscriberGroup(ctx, 0, subscriberGroup)
+	return session.v1StreamManagerTopics.destroySubscriberGroup(ctx, proxyID, subscriberGroup)
 }
 
 // ensurePublisher ensures a publisher.
@@ -636,8 +665,8 @@ func (m *streamManagerV1) ensurePublisher(ctx context.Context, topicName string,
 }
 
 // ensureSubscriber ensures a subscriber.
-func (m *streamManagerV1) ensureSubscriber(ctx context.Context, topicName string, subscriberGroup *string, binFilter []byte) (*subscriber.EnsureSubscriberResult, error) {
-	req, err := m.newEnsureSubscriberRequest(topicName, subscriberGroup, binFilter)
+func (m *streamManagerV1) ensureSubscriber(ctx context.Context, topicName string, options *subscriber.Options) (*subscriber.EnsureSubscriberResult, error) {
+	req, err := m.newEnsureSubscriberRequest(topicName, options)
 	if err != nil {
 		return nil, err
 	}
@@ -672,14 +701,26 @@ func (m *streamManagerV1) ensureSubscriber(ctx context.Context, topicName string
 	return &s, nil
 }
 
-// receive calls receive for a subscriber.
-func (m *streamManagerV1) receive(ctx context.Context, proxyID int32, maxMessages *int32) ([]*pb1topics.TopicElement, error) {
-	req, err := m.newSubscriberReceiveRequest(proxyID, maxMessages)
+// peekOrReceive calls peek or receive for a subscriber.
+func (m *streamManagerV1) peekOrReceive(ctx context.Context, proxyID int32, maxMessages *int32, isReceive bool) ([]*pb1topics.TopicElement, error) {
+	var (
+		messageType pb1topics.TopicServiceRequestType
+		err         error
+		req         *pb1.ProxyRequest
+	)
+	if isReceive {
+		messageType = pb1topics.TopicServiceRequestType_Receive
+		req, err = m.newSubscriberReceiveRequest(proxyID, maxMessages)
+	} else {
+		messageType = pb1topics.TopicServiceRequestType_PeekAtPosition
+		req, err = m.newSubscriberPeekRequest(proxyID, maxMessages)
+	}
+
 	if err != nil {
 		return nil, err
 	}
 
-	requestType, err := m.submitTopicRequest(req, pb1topics.TopicServiceRequestType_SimpleReceive)
+	requestType, err := m.submitTopicRequest(req, messageType)
 
 	newCtx, cancel := m.session.ensureContext(ctx)
 	if cancel != nil {
@@ -789,7 +830,7 @@ func (m *streamManagerV1) ensureSubscriberGroup(ctx context.Context, topicName s
 	return err
 }
 
-// destroySubscriberGroup destroys a subscriber.
+// destroySubscriberGroup destroys a subscriber group.
 func (m *streamManagerV1) destroySubscriberGroup(ctx context.Context, proxyID int32, subscriberGroup string) error {
 	req, err := m.newDestroySubscriberGroupRequest(proxyID, subscriberGroup)
 	if err != nil {
@@ -952,7 +993,7 @@ func ensureV1StreamManagerTopics(session *Session) error {
 }
 
 func releaseTopicInternal[V any](ctx context.Context, bt *baseTopicsClient[V], destroy bool) error {
-	if bt.isDestroyed || bt.isReleased {
+	if bt.isTopicDestroyed() || bt.isTopicReleased() {
 		return ErrTopicDestroyedOrReleased
 	}
 
@@ -969,12 +1010,12 @@ func releaseTopicInternal[V any](ctx context.Context, bt *baseTopicsClient[V], d
 		if err != nil {
 			return err
 		}
-		bt.isDestroyed = true
+		bt.setDestroyed()
 		bt.generateTopicLifecycleEvent(nil, TopicDestroyed)
 	} else {
 		if existingTopic, ok := bt.session.topics[bt.name]; ok {
 			bt.generateTopicLifecycleEvent(existingTopic, TopicReleased)
-			bt.isReleased = true
+			bt.setReleased()
 		}
 	}
 
@@ -1009,11 +1050,25 @@ func (m *streamManagerV1) newDestroyPublisherOrSubscriberRequest(proxyID int32,
 	return m.newWrapperProxyTopicsRequest("", requestType, anyReq)
 }
 
-func (m *streamManagerV1) newEnsureSubscriberRequest(topicName string, subscriberGroup *string, binFilter []byte) (*pb1.ProxyRequest, error) {
+func (m *streamManagerV1) newEnsureSubscriberRequest(topicName string, options *subscriber.Options) (*pb1.ProxyRequest, error) {
+	var (
+		err       error
+		binFilter []byte
+	)
+
+	if options.Filter != nil {
+		binFilter, err = m.session.genericSerializer.Serialize(options.Filter)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	req := &pb1topics.EnsureSimpleSubscriberRequest{
 		Topic:           topicName,
 		Filter:          binFilter,
-		SubscriberGroup: subscriberGroup,
+		SubscriberGroup: options.SubscriberGroup,
+		Extractor:       options.Extractor,
+		Channels:        options.Channels,
 	}
 
 	anyReq, err := anypb.New(req)
@@ -1034,6 +1089,18 @@ func (m *streamManagerV1) newSubscriberReceiveRequest(ProxyID int32, maxMessages
 	}
 	return m.newWrapperProxyPublisherRequest(ProxyID, pb1topics.TopicServiceRequestType_SimpleReceive, anyReq)
 }
+func (m *streamManagerV1) newSubscriberPeekRequest(ProxyID int32, maxMessages *int32) (*pb1.ProxyRequest, error) {
+	req := &pb1topics.SimpleReceiveRequest{
+		MaxMessages: maxMessages,
+	}
+
+	anyReq, err := anypb.New(req)
+	if err != nil {
+		return nil, err
+	}
+	// TODO: FIX THIS when SImplePeek available
+	return m.newWrapperProxyPublisherRequest(ProxyID, pb1topics.TopicServiceRequestType_SimpleReceive, anyReq)
+}
 
 func (m *streamManagerV1) newSubscriberCommitRequest(ProxyID int32, channel int32, position *pb1topics.TopicPosition) (*pb1.ProxyRequest, error) {
 	req := &pb1topics.ChannelAndPosition{

coherence/topics_events.go

@@ -99,7 +99,7 @@ func (t *topicLifecycleListener[V]) on(event TopicLifecycleEventType, callback f
 }
 
 func (bt *baseTopicsClient[V]) AddLifecycleListener(listener TopicLifecycleListener[V]) error {
-	if bt.isDestroyed || bt.isReleased {
+	if bt.isTopicDestroyed() || bt.isTopicReleased() {
 		return ErrTopicDestroyedOrReleased
 	}
 
@@ -108,7 +108,7 @@ func (bt *baseTopicsClient[V]) AddLifecycleListener(listener TopicLifecycleListe
 }
 
 func (bt *baseTopicsClient[V]) RemoveLifecycleListener(listener TopicLifecycleListener[V]) error {
-	if bt.isDestroyed || bt.isReleased {
+	if bt.isTopicDestroyed() || bt.isTopicReleased() {
 		return ErrTopicDestroyedOrReleased
 	}
 

coherence/v1client.go

@@ -37,6 +37,7 @@ var (
 	ALL     logLevel = 5 // all messages
 
 	// current log level
+	logLevelMutex   sync.RWMutex
 	currentLogLevel int
 )
 
@@ -251,6 +252,9 @@ func (m *streamManagerV1) processResponseMessage(id int64, resp *responseMessage
 
 // logMessage logs a message only if the level <= currentLogLevel
 func logMessage(level logLevel, format string, args ...any) {
+	logLevelMutex.RLock()
+	defer logLevelMutex.RUnlock()
+
 	if int(level) <= currentLogLevel {
 		log.Println(getLogMessage(level, format, args...))
 	}
@@ -561,7 +565,7 @@ func (m *streamManagerV1) ensure(ctx context.Context, name string, IDMap safeMap
 		return nil, err
 	}
 
-	// store the cache id in the session , no lock required as already locked
+	// store the cache/topic/queue id in the session, no lock required as already locked
 	ID = &message.Value
 	IDMap.Add(name, *ID)
 

java/coherence-go-test/src/main/resources/test-cache-config.xml

@@ -32,6 +32,15 @@
       <topic-name>*</topic-name>
       <scheme-name>topic-scheme</scheme-name>
     </topic-mapping>
+    <topic-mapping>
+      <topic-name>durable-*</topic-name>
+      <scheme-name>topic-scheme</scheme-name>
+      <subscriber-groups>
+        <subscriber-group>
+          <name>subscriber-group</name>
+        </subscriber-group>
+      </subscriber-groups>
+    </topic-mapping>
   </topic-scheme-mapping>
 
   <caching-schemes>