Skip to content

Commit 35f6366

Browse files
expliseexplise
committed
feat(go): add diagnostic event subscriptions to TCP client
Introduce a DiagnosticEvent enum (shutdown / disconnected / connected / signed_in / signed_out) mirroring the Rust SDK, and expose Client.SubscribeEvents so callers can observe client-level lifecycle changes. The TCP client wires events into connect, disconnect, shutdown, login and logout paths via a non-blocking fan-out broadcaster: each subscriber gets an independent buffered channel, and a slow subscriber drops its oldest event rather than blocking the publisher. Channels are closed after the final shutdown event is delivered. Includes unit tests for the contract and broadcaster plus an integration test covering the end-to-end event flow.
1 parent 6b908ca commit 35f6366

7 files changed

Lines changed: 560 additions & 5 deletions

File tree

‎foreign/go/client/tcp/tcp_core.go‎

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@ type IggyTcpClient struct {
6060
currentServerAddress string
6161
connectedAt time.Time
6262
state iggcon.State
63+
events *eventBroadcaster
6364
// respHeader is the reused response-status read buffer; guarded by c.mtx.
6465
respHeader [ResponseInitialBytesLength]byte
6566
}
@@ -220,6 +221,7 @@ func NewIggyTcpClient(logger *slog.Logger, options ...Option) *IggyTcpClient {
220221
connectedAt: time.Time{},
221222
leaderRedirectionState: iggcon.LeaderRedirectionState{},
222223
currentServerAddress: opts.config.serverAddress,
224+
events: newEventBroadcaster(),
223225
}
224226
}
225227

@@ -548,7 +550,7 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error {
548550
if !c.config.reconnection.enabled {
549551
c.logger.Warn("Automatic reconnection is disabled.")
550552
}
551-
// TODO publish event disconnected
553+
c.events.publish(iggcon.DiagnosticEventDisconnected)
552554
return err
553555
}
554556

@@ -558,6 +560,7 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error {
558560
c.connectedAt = time.Now()
559561
c.logger.Info("Iggy client has connected to the Iggy server", slog.String("client_address", c.clientAddress), slog.String("server_address", c.currentServerAddress))
560562
c.mtx.Unlock()
563+
c.events.publish(iggcon.DiagnosticEventConnected)
561564
return nil
562565
}
563566

@@ -624,7 +627,7 @@ func (c *IggyTcpClient) disconnect() error {
624627
}
625628

626629
c.logger.Info("Iggy client has disconnected from server.", slog.String("client_address", c.clientAddress))
627-
// TODO event pushing logic
630+
c.events.publish(iggcon.DiagnosticEventDisconnected)
628631
return nil
629632
}
630633

@@ -646,10 +649,17 @@ func (c *IggyTcpClient) shutdown() error {
646649

647650
c.state = iggcon.StateShutdown
648651
c.logger.Info("Iggy TCP client has been shutdown.", slog.String("client_address", c.clientAddress))
649-
// TODO push shutdown event
652+
c.events.publish(iggcon.DiagnosticEventShutdown)
653+
c.events.close()
650654
return nil
651655
}
652656

653657
func (c *IggyTcpClient) Close() error {
654658
return c.shutdown()
655659
}
660+
661+
// SubscribeEvents returns an independent channel of client lifecycle events.
662+
// The channel is closed on shutdown, after the final DiagnosticEventShutdown.
663+
func (c *IggyTcpClient) SubscribeEvents() <-chan iggcon.DiagnosticEvent {
664+
return c.events.subscribe()
665+
}

‎foreign/go/client/tcp/tcp_events.go‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package tcp
19+
20+
import (
21+
"sync"
22+
23+
iggcon "github.com/apache/iggy/foreign/go/contracts"
24+
)
25+
26+
// subscriberBufferSize bounds each subscriber's channel. A subscriber that
27+
// fails to drain within this many events will start dropping the oldest
28+
// events; the publisher is never blocked.
29+
const subscriberBufferSize = 1000
30+
31+
// eventBroadcaster fans out diagnostic events to any number of subscribers.
32+
// Each call to subscribe returns an independent channel; publish delivers
33+
// the event to every live subscriber non-blockingly. close releases all
34+
// subscriber channels and is idempotent.
35+
type eventBroadcaster struct {
36+
mu sync.Mutex
37+
subscribers []chan iggcon.DiagnosticEvent
38+
closed bool
39+
}
40+
41+
func newEventBroadcaster() *eventBroadcaster {
42+
return &eventBroadcaster{}
43+
}
44+
45+
func (b *eventBroadcaster) subscribe() <-chan iggcon.DiagnosticEvent {
46+
b.mu.Lock()
47+
defer b.mu.Unlock()
48+
49+
ch := make(chan iggcon.DiagnosticEvent, subscriberBufferSize)
50+
if b.closed {
51+
close(ch)
52+
return ch
53+
}
54+
b.subscribers = append(b.subscribers, ch)
55+
return ch
56+
}
57+
58+
func (b *eventBroadcaster) publish(event iggcon.DiagnosticEvent) {
59+
b.mu.Lock()
60+
defer b.mu.Unlock()
61+
62+
if b.closed {
63+
return
64+
}
65+
for _, ch := range b.subscribers {
66+
select {
67+
case ch <- event:
68+
default:
69+
// Subscriber is not draining fast enough. Drop the oldest event
70+
// to make room for the newest, preferring recency. Keeps the
71+
// publisher non-blocking even with a slow subscriber.
72+
select {
73+
case <-ch:
74+
default:
75+
}
76+
select {
77+
case ch <- event:
78+
default:
79+
}
80+
}
81+
}
82+
}
83+
84+
func (b *eventBroadcaster) close() {
85+
b.mu.Lock()
86+
defer b.mu.Unlock()
87+
88+
if b.closed {
89+
return
90+
}
91+
b.closed = true
92+
for _, ch := range b.subscribers {
93+
close(ch)
94+
}
95+
b.subscribers = nil
96+
}

‎foreign/go/client/tcp/tcp_events_test.go‎

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package tcp
19+
20+
import (
21+
"testing"
22+
"time"
23+
24+
iggcon "github.com/apache/iggy/foreign/go/contracts"
25+
)
26+
27+
func recvWithTimeout(t *testing.T, ch <-chan iggcon.DiagnosticEvent) (iggcon.DiagnosticEvent, bool) {
28+
t.Helper()
29+
select {
30+
case ev, ok := <-ch:
31+
return ev, ok
32+
case <-time.After(time.Second):
33+
t.Fatal("timed out waiting for event")
34+
return 0, false
35+
}
36+
}
37+
38+
func TestEventBroadcaster_DeliversToAllSubscribers(t *testing.T) {
39+
b := newEventBroadcaster()
40+
a := b.subscribe()
41+
c := b.subscribe()
42+
43+
b.publish(iggcon.DiagnosticEventConnected)
44+
45+
if ev, _ := recvWithTimeout(t, a); ev != iggcon.DiagnosticEventConnected {
46+
t.Errorf("subscriber A got %v, want connected", ev)
47+
}
48+
if ev, _ := recvWithTimeout(t, c); ev != iggcon.DiagnosticEventConnected {
49+
t.Errorf("subscriber B got %v, want connected", ev)
50+
}
51+
}
52+
53+
func TestEventBroadcaster_PreservesOrderForOneSubscriber(t *testing.T) {
54+
b := newEventBroadcaster()
55+
ch := b.subscribe()
56+
57+
want := []iggcon.DiagnosticEvent{
58+
iggcon.DiagnosticEventConnected,
59+
iggcon.DiagnosticEventSignedIn,
60+
iggcon.DiagnosticEventDisconnected,
61+
}
62+
for _, ev := range want {
63+
b.publish(ev)
64+
}
65+
for i, w := range want {
66+
got, _ := recvWithTimeout(t, ch)
67+
if got != w {
68+
t.Errorf("event %d: got %v, want %v", i, got, w)
69+
}
70+
}
71+
}
72+
73+
func TestEventBroadcaster_CloseDeliversNoMoreEvents(t *testing.T) {
74+
b := newEventBroadcaster()
75+
ch := b.subscribe()
76+
77+
b.close()
78+
79+
// channel should be closed; receive returns zero, ok=false
80+
select {
81+
case _, ok := <-ch:
82+
if ok {
83+
t.Fatal("expected channel to be closed")
84+
}
85+
case <-time.After(time.Second):
86+
t.Fatal("timed out waiting for closed channel")
87+
}
88+
89+
// subsequent publish is a no-op
90+
b.publish(iggcon.DiagnosticEventConnected)
91+
92+
// late subscribe returns an already-closed channel
93+
late := b.subscribe()
94+
select {
95+
case _, ok := <-late:
96+
if ok {
97+
t.Fatal("expected late-subscribe channel to be closed")
98+
}
99+
case <-time.After(time.Second):
100+
t.Fatal("timed out waiting for closed late-subscribe channel")
101+
}
102+
}
103+
104+
func TestEventBroadcaster_SlowSubscriberDoesNotBlockPublisher(t *testing.T) {
105+
b := newEventBroadcaster()
106+
_ = b.subscribe() // never drained
107+
108+
// Publish more events than the buffer; if the publisher blocked or
109+
// panicked we'd hang or fail here.
110+
for range subscriberBufferSize * 2 {
111+
b.publish(iggcon.DiagnosticEventConnected)
112+
}
113+
}
114+
115+
func TestEventBroadcaster_CloseIsIdempotent(t *testing.T) {
116+
b := newEventBroadcaster()
117+
b.subscribe()
118+
b.close()
119+
b.close() // must not panic
120+
}

‎foreign/go/client/tcp/tcp_session_management.go‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ func (c *IggyTcpClient) LoginUser(ctx context.Context, username string, password
4141

4242
c.logger.Info("Iggy client has signed in successfully.", slog.String("client_address", c.clientAddress))
4343
identity := binaryserialization.DeserializeLogInResponse(buffer)
44+
c.events.publish(iggcon.DiagnosticEventSignedIn)
4445
shouldRedirect, err := c.HandleLeaderRedirection(ctx)
4546
if err != nil {
4647
return nil, err
@@ -65,6 +66,7 @@ func (c *IggyTcpClient) LoginWithPersonalAccessToken(ctx context.Context, token
6566

6667
c.logger.Info("Iggy client has signed in successfully.", slog.String("client_address", c.clientAddress))
6768
identity := binaryserialization.DeserializeLogInResponse(buffer)
69+
c.events.publish(iggcon.DiagnosticEventSignedIn)
6870
shouldRedirect, err := c.HandleLeaderRedirection(ctx)
6971
if err != nil {
7072
return nil, err
@@ -79,8 +81,11 @@ func (c *IggyTcpClient) LoginWithPersonalAccessToken(ctx context.Context, token
7981
}
8082

8183
func (c *IggyTcpClient) LogoutUser(ctx context.Context) error {
82-
_, err := c.do(ctx, &command.LogoutUser{})
83-
return err
84+
if _, err := c.do(ctx, &command.LogoutUser{}); err != nil {
85+
return err
86+
}
87+
c.events.publish(iggcon.DiagnosticEventSignedOut)
88+
return nil
8489
}
8590

8691
func (c *IggyTcpClient) HandleLeaderRedirection(ctx context.Context) (bool, error) {

‎foreign/go/contracts/client.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,10 @@ type Client interface {
2929
// GetConnectionInfo returns the current connection information including protocol and server address
3030
GetConnectionInfo() *ConnectionInfo
3131

32+
// SubscribeEvents returns an independent channel of client lifecycle events.
33+
// The channel is closed on shutdown, after the final DiagnosticEventShutdown.
34+
SubscribeEvents() <-chan DiagnosticEvent
35+
3236
// GetClusterMetadata get the metadata of the cluster including node information, roles, and status.
3337
// Authentication is required.
3438
GetClusterMetadata(ctx context.Context) (*ClusterMetadata, error)

‎foreign/go/contracts/diagnostic_event.go‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package iggcon
19+
20+
// DiagnosticEvent describes a client-level lifecycle change that subscribers
21+
// can observe via Client.SubscribeEvents.
22+
type DiagnosticEvent int
23+
24+
const (
25+
DiagnosticEventShutdown DiagnosticEvent = iota
26+
DiagnosticEventDisconnected
27+
DiagnosticEventConnected
28+
DiagnosticEventSignedIn
29+
DiagnosticEventSignedOut
30+
)
31+
32+
func (e DiagnosticEvent) String() string {
33+
switch e {
34+
case DiagnosticEventShutdown:
35+
return "shutdown"
36+
case DiagnosticEventDisconnected:
37+
return "disconnected"
38+
case DiagnosticEventConnected:
39+
return "connected"
40+
case DiagnosticEventSignedIn:
41+
return "signed_in"
42+
case DiagnosticEventSignedOut:
43+
return "signed_out"
44+
default:
45+
return "unknown"
46+
}
47+
}

0 commit comments

Comments
 (0)