initial subscriber events

Author: Tim Middleton

coherence/topics.go

@@ -87,6 +87,12 @@ type Publisher[V any] interface {
 	// Close closes a publisher and releases all resources associated with it in the client
 	// and on the server.
 	Close(ctx context.Context) error
+
+	// AddLifecycleListener adds a [PublisherLifecycleListener] to this topic.
+	AddLifecycleListener(listener PublisherLifecycleListener[V]) error
+
+	// RemoveLifecycleListener removes a [PublisherLifecycleListener] from this topic.
+	RemoveLifecycleListener(listener PublisherLifecycleListener[V]) error
 }
 
 // Subscriber subscribes directly to a [NamedTopic], or to a subscriber group of a [NamedTopic].
@@ -167,7 +173,7 @@ type topicPublisher[V any] struct {
 	channelCount         int32
 	options              *publisher.Options
 	valueSerializer      Serializer[V]
-	mutex                *sync.RWMutex
+	mutex                sync.RWMutex
 	lifecycleListenersV1 []*PublisherLifecycleListener[V]
 }
 

test/e2e/topics/publisher_event_test.go

@@ -0,0 +1,104 @@
+/*
+ * Copyright (c) 2025 Oracle and/or its affiliates.
+ * Licensed under the Universal Permissive License v 1.0 as shown at
+ * https://oss.oracle.com/licenses/upl.
+ */
+
+package topics
+
+import (
+	"context"
+	"github.com/onsi/gomega"
+	"github.com/oracle/coherence-go-client/v2/coherence"
+	"github.com/oracle/coherence-go-client/v2/test/utils"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestPublisherEventsDestroyOnServer(t *testing.T) {
+	var (
+		g   = gomega.NewWithT(t)
+		err error
+	)
+
+	const topicName = "publisher-events"
+
+	t.Setenv("COHERENCE_LOG_LEVEL", "ALL")
+
+	session1, topic1 := getSessionAndTopic[string](g, topicName)
+	defer session1.Close()
+
+	pub, err := topic1.CreatePublisher(context.Background())
+	g.Expect(err).Should(gomega.Not(gomega.HaveOccurred()))
+
+	utils.Sleep(5)
+
+	listener := NewCountingPublisherListener[string](topicName)
+
+	g.Expect(pub.AddLifecycleListener(listener.listener)).ShouldNot(gomega.HaveOccurred())
+
+	_, err = utils.IssueGetRequest(utils.GetTestContext().RestURL + "/destroyTopic/" + topicName)
+	g.Expect(err).Should(gomega.Not(gomega.HaveOccurred()))
+
+	utils.Sleep(5)
+
+	g.Eventually(func() int32 { return listener.destroyCount }).
+		WithTimeout(10 * time.Second).Should(gomega.Equal(int32(1)))
+
+	g.Expect(topic1.Close(context.Background())).Should(gomega.Equal(coherence.ErrTopicDestroyedOrReleased))
+}
+
+func TestPublisherEventsDestroyByClient(t *testing.T) {
+	var (
+		g   = gomega.NewWithT(t)
+		ctx = context.Background()
+	)
+
+	const topicName = "publisher-events-client-destroy"
+
+	t.Setenv("COHERENCE_LOG_LEVEL", "ALL")
+
+	session1, topic1 := getSessionAndTopic[string](g, topicName)
+	defer session1.Close()
+
+	pub, err := topic1.CreatePublisher(context.Background())
+	g.Expect(err).Should(gomega.Not(gomega.HaveOccurred()))
+
+	utils.Sleep(5)
+
+	listener := NewCountingPublisherListener[string](topicName)
+
+	g.Expect(pub.AddLifecycleListener(listener.listener)).ShouldNot(gomega.HaveOccurred())
+
+	g.Expect(pub.Close(ctx)).ShouldNot(gomega.HaveOccurred())
+	utils.Sleep(1)
+
+	g.Eventually(func() int32 { return listener.releaseCount }).
+		WithTimeout(10 * time.Second).Should(gomega.Equal(int32(1)))
+}
+
+func NewCountingPublisherListener[V any](name string) *CountingPublisherListener[V] {
+	countingListener := CountingPublisherListener[V]{
+		name:     name,
+		listener: coherence.NewPublisherLifecycleListener[V](),
+	}
+
+	countingListener.listener.OnDestroyed(func(_ coherence.PublisherLifecycleEvent[V]) {
+		atomic.AddInt32(&countingListener.destroyCount, 1)
+	}).OnReleased(func(_ coherence.PublisherLifecycleEvent[V]) {
+		atomic.AddInt32(&countingListener.releaseCount, 1)
+	}).OnAny(func(_ coherence.PublisherLifecycleEvent[V]) {
+		atomic.AddInt32(&countingListener.allCount, 1)
+	})
+
+	return &countingListener
+}
+
+type CountingPublisherListener[V any] struct {
+	listener     coherence.PublisherLifecycleListener[V]
+	name         string
+	releaseCount int32
+	destroyCount int32
+	allCount     int32
+}