feat(go): add diagnostic event subscriptions to TCP client

foreign/go/client/tcp/tcp_core.go

@@ -91,6 +91,9 @@ type IggyTcpClient struct {
 	connectedAt            time.Time
 	transportState         iggcon.TransportState
 	sessionState           iggcon.SessionState
+	// events fans out client lifecycle events to any number of subscribers;
+	// guarded by its own mutex.
+	events eventBroadcaster
 	// session carries the consensus client identity and request watermark;
 	// guarded by c.mtx.
 	session *vsr.Session
@@ -296,6 +299,7 @@ func NewIggyTcpClient(logger *slog.Logger, options ...Option) *IggyTcpClient {
 		conn:                   nil,
 		transportState:         iggcon.TransportStateDisconnected,
 		sessionState:           iggcon.SessionStateUnauthenticated,
+		events:                 eventBroadcaster{},
 		connectedAt:            time.Time{},
 		leaderRedirectionState: iggcon.LeaderRedirectionState{},
 		currentServerAddress:   opts.config.serverAddress,
@@ -944,10 +948,15 @@ func (c *IggyTcpClient) waitBeforeReplay(ctx context.Context, deadline time.Time
 	}
 }
 
-// invalidateConnLocked closes the connection and marks it as disconnected
+// invalidateConnLocked closes the connection and marks it as disconnected,
+// publishing a Disconnected event when the transport actually leaves a live
+// state.
 func (c *IggyTcpClient) invalidateConnLocked() {
 	_ = c.closeConnLocked()
-	c.transportState = iggcon.TransportStateDisconnected
+	if c.transportState != iggcon.TransportStateDisconnected {
+		c.transportState = iggcon.TransportStateDisconnected
+		c.events.publish(iggcon.DiagnosticEventDisconnected)
+	}
 	c.sessionState = iggcon.SessionStateUnauthenticated
 	c.session.Reset()
 	c.groups.clear()
@@ -1046,6 +1055,7 @@ func (c *IggyTcpClient) Connect(ctx context.Context) (err error) {
 		// connected.
 		c.mtx.Lock()
 		c.transportState = iggcon.TransportStateDisconnected
+		c.events.publish(iggcon.DiagnosticEventDisconnected)
 		c.mtx.Unlock()
 		c.logger.Error("No server address to connect to.")
 		return ierror.ErrCannotEstablishConnection
@@ -1117,12 +1127,14 @@ func (c *IggyTcpClient) Connect(ctx context.Context) (err error) {
 			return lastErr
 		}); err != nil {
 		c.mtx.Lock()
-		c.transportState = iggcon.TransportStateDisconnected
+		if c.transportState != iggcon.TransportStateDisconnected {
+			c.transportState = iggcon.TransportStateDisconnected
+			c.events.publish(iggcon.DiagnosticEventDisconnected)
+		}
 		c.mtx.Unlock()
 		if !c.config.reconnection.enabled {
 			c.logger.Warn("Automatic reconnection is disabled.")
 		}
-		// TODO publish event disconnected
 		return err
 	}
 
@@ -1153,6 +1165,7 @@ func (c *IggyTcpClient) Connect(ctx context.Context) (err error) {
 	// The server fence does not survive the old socket, so the new connection
 	// starts from a fresh client identity.
 	c.session.Reset()
+	c.events.publish(iggcon.DiagnosticEventConnected)
 	clientAddress := c.clientAddress
 	serverAddress := c.currentServerAddress
 	c.mtx.Unlock()
@@ -1452,7 +1465,7 @@ func (c *IggyTcpClient) disconnectLocked() error {
 	err := c.closeConnLocked()
 
 	c.logger.Info("Iggy client has disconnected from server.", slog.String("client_address", c.clientAddress))
-	// TODO event pushing logic
+	c.events.publish(iggcon.DiagnosticEventDisconnected)
 	return err
 }
 
@@ -1482,10 +1495,19 @@ func (c *IggyTcpClient) shutdown() error {
 	c.groups.clear()
 	c.topics.clearCounts()
 	c.logger.Info("Iggy TCP client has been shutdown.", slog.String("client_address", c.clientAddress))
-	// TODO push shutdown event
+	c.events.publish(iggcon.DiagnosticEventShutdown)
+	c.events.close()
 	return err
 }
 
 func (c *IggyTcpClient) Close() error {
 	return c.shutdown()
 }
+
+// SubscribeEvents returns an independent channel of client lifecycle events
+// and an unsubscribe function that removes the subscription and closes the
+// channel. The channel is also closed on shutdown, after the final
+// DiagnosticEventShutdown.
+func (c *IggyTcpClient) SubscribeEvents() (<-chan iggcon.DiagnosticEvent, func()) {
+	return c.events.subscribe()
+}

foreign/go/client/tcp/tcp_events.go

@@ -0,0 +1,106 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package tcp
+
+import (
+	"slices"
+	"sync"
+
+	iggcon "github.com/apache/iggy/foreign/go/contracts"
+)
+
+// subscriberBufferSize bounds each subscriber's channel. A subscriber that
+// fails to drain within this many events will start dropping the oldest
+// events; the publisher is never blocked.
+const subscriberBufferSize = 1000
+
+// eventBroadcaster fans out diagnostic events to any number of subscribers.
+// Each call to subscribe returns an independent channel; publish delivers
+// the event to every live subscriber non-blockingly. close releases all
+// subscriber channels and is idempotent.
+type eventBroadcaster struct {
+	mu          sync.Mutex
+	subscribers []chan iggcon.DiagnosticEvent
+	closed      bool
+}
+
+// subscribe returns a new subscriber channel and an unsubscribe function that
+// is idempotent and safe to call after the broadcaster is closed.
+func (b *eventBroadcaster) subscribe() (<-chan iggcon.DiagnosticEvent, func()) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	ch := make(chan iggcon.DiagnosticEvent, subscriberBufferSize)
+	if b.closed {
+		close(ch)
+		return ch, func() {}
+	}
+	b.subscribers = append(b.subscribers, ch)
+
+	unsubscribe := func() {
+		b.mu.Lock()
+		defer b.mu.Unlock()
+		for i, sub := range b.subscribers {
+			if sub == ch {
+				b.subscribers = slices.Delete(b.subscribers, i, i+1)
+				close(ch)
+				return
+			}
+		}
+	}
+	return ch, unsubscribe
+}
+
+func (b *eventBroadcaster) publish(event iggcon.DiagnosticEvent) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if b.closed {
+		return
+	}
+	for _, ch := range b.subscribers {
+		select {
+		case ch <- event:
+		default:
+			// Subscriber is not draining fast enough. Drop the oldest event
+			// to make room for the newest, preferring recency. Keeps the
+			// publisher non-blocking even with a slow subscriber.
+			select {
+			case <-ch:
+			default:
+			}
+			// The drained slot above guarantees capacity, so this send
+			// cannot block.
+			ch <- event
+		}
+	}
+}
+
+func (b *eventBroadcaster) close() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if b.closed {
+		return
+	}
+	b.closed = true
+	for _, ch := range b.subscribers {
+		close(ch)
+	}
+	b.subscribers = nil
+}

foreign/go/client/tcp/tcp_events_test.go

@@ -0,0 +1,149 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package tcp
+
+import (
+	"testing"
+	"time"
+
+	iggcon "github.com/apache/iggy/foreign/go/contracts"
+)
+
+func recvWithTimeout(t *testing.T, ch <-chan iggcon.DiagnosticEvent) (iggcon.DiagnosticEvent, bool) {
+	t.Helper()
+	select {
+	case ev, ok := <-ch:
+		return ev, ok
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for event")
+		return 0, false
+	}
+}
+
+func TestEventBroadcaster_DeliversToAllSubscribers(t *testing.T) {
+	b := eventBroadcaster{}
+	a, _ := b.subscribe()
+	c, _ := b.subscribe()
+
+	b.publish(iggcon.DiagnosticEventConnected)
+
+	if ev, _ := recvWithTimeout(t, a); ev != iggcon.DiagnosticEventConnected {
+		t.Errorf("subscriber A got %v, want connected", ev)
+	}
+	if ev, _ := recvWithTimeout(t, c); ev != iggcon.DiagnosticEventConnected {
+		t.Errorf("subscriber B got %v, want connected", ev)
+	}
+}
+
+func TestEventBroadcaster_PreservesOrderForOneSubscriber(t *testing.T) {
+	b := eventBroadcaster{}
+	ch, _ := b.subscribe()
+
+	want := []iggcon.DiagnosticEvent{
+		iggcon.DiagnosticEventConnected,
+		iggcon.DiagnosticEventSignedIn,
+		iggcon.DiagnosticEventDisconnected,
+	}
+	for _, ev := range want {
+		b.publish(ev)
+	}
+	for i, w := range want {
+		got, _ := recvWithTimeout(t, ch)
+		if got != w {
+			t.Errorf("event %d: got %v, want %v", i, got, w)
+		}
+	}
+}
+
+func TestEventBroadcaster_CloseDeliversNoMoreEvents(t *testing.T) {
+	b := eventBroadcaster{}
+	ch, _ := b.subscribe()
+
+	b.close()
+
+	// channel should be closed; receive returns zero, ok=false
+	select {
+	case _, ok := <-ch:
+		if ok {
+			t.Fatal("expected channel to be closed")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for closed channel")
+	}
+
+	// subsequent publish is a no-op
+	b.publish(iggcon.DiagnosticEventConnected)
+
+	// late subscribe returns an already-closed channel
+	late, _ := b.subscribe()
+	select {
+	case _, ok := <-late:
+		if ok {
+			t.Fatal("expected late-subscribe channel to be closed")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for closed late-subscribe channel")
+	}
+}
+
+func TestEventBroadcaster_SlowSubscriberDoesNotBlockPublisher(t *testing.T) {
+	b := eventBroadcaster{}
+	_, _ = b.subscribe() // never drained
+
+	// Publish more events than the buffer; if the publisher blocked or
+	// panicked we'd hang or fail here.
+	for range subscriberBufferSize * 2 {
+		b.publish(iggcon.DiagnosticEventConnected)
+	}
+}
+
+func TestEventBroadcaster_CloseIsIdempotent(t *testing.T) {
+	b := eventBroadcaster{}
+	b.subscribe()
+	b.close()
+	b.close() // must not panic
+}
+
+func TestEventBroadcaster_UnsubscribeStopsDelivery(t *testing.T) {
+	b := eventBroadcaster{}
+	ch, unsubscribe := b.subscribe()
+
+	unsubscribe()
+
+	if _, ok := <-ch; ok {
+		t.Fatal("expected channel to be closed after unsubscribe")
+	}
+	b.publish(iggcon.DiagnosticEventConnected)
+
+	other, _ := b.subscribe()
+	b.publish(iggcon.DiagnosticEventSignedIn)
+	if ev, _ := recvWithTimeout(t, other); ev != iggcon.DiagnosticEventSignedIn {
+		t.Errorf("remaining subscriber got %v, want signed_in", ev)
+	}
+}
+
+func TestEventBroadcaster_UnsubscribeIsIdempotentAndSafeAfterClose(t *testing.T) {
+	b := eventBroadcaster{}
+	_, unsubscribe := b.subscribe()
+
+	unsubscribe()
+	unsubscribe() // second call must not panic or double-close
+
+	b.close()
+	unsubscribe() // after close must not panic
+}

foreign/go/client/tcp/tcp_session_management.go

@@ -128,6 +128,7 @@ func (c *IggyTcpClient) signIn(ctx context.Context, code uint32, body []byte) (*
 	if err == nil {
 		c.sessionState = iggcon.SessionStateAuthenticated
 		c.loggedOut = false
+		c.events.publish(iggcon.DiagnosticEventSignedIn)
 	} else {
 		// The server committed a Register this client failed to adopt, so the
 		// connection carries a session the local state does not track. It is
@@ -251,6 +252,7 @@ func (c *IggyTcpClient) LogoutUser(ctx context.Context) error {
 	c.loggedOut = true
 	c.groups.clear()
 	c.topics.clearCounts()
+	c.events.publish(iggcon.DiagnosticEventSignedOut)
 	c.mtx.Unlock()
 	c.forgetLogin()
 	return nil
@@ -328,6 +330,10 @@ func (c *IggyTcpClient) redirectToLeader(ctx context.Context, generation uint64)
 	c.currentServerAddress = leaderAddress
 	c.mtx.Unlock()
 
+	// Published after the teardown so a subscriber sees the redirect on the
+	// same connection generation it decided on.
+	c.events.publish(iggcon.DiagnosticEventRedirected)
+
 	return true, nil
 }
 

foreign/go/contracts/client.go

@@ -29,6 +29,12 @@ type Client interface {
 	// GetConnectionInfo returns the current connection information including protocol and server address
 	GetConnectionInfo() *ConnectionInfo
 
+	// SubscribeEvents returns an independent channel of client lifecycle events
+	// and an unsubscribe function that removes the subscription and closes the
+	// channel. The channel is also closed on shutdown, after the final
+	// DiagnosticEventShutdown.
+	SubscribeEvents() (<-chan DiagnosticEvent, func())
+
 	// GetClusterMetadata get the metadata of the cluster including node information, roles, and status.
 	// Authentication is required.
 	GetClusterMetadata(ctx context.Context) (*ClusterMetadata, error)