Skip to content

Commit 43d079a

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 43d079a

7 files changed

Lines changed: 624 additions & 7 deletions

File tree

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

Lines changed: 20 additions & 5 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

@@ -431,12 +433,16 @@ func (c *IggyTcpClient) sendLocked(wirePayload []byte) ([]byte, error) {
431433
return buffer, nil
432434
}
433435

434-
// invalidateConnLocked closes the connection and marks it as disconnected
436+
// invalidateConnLocked closes the connection and marks it as disconnected,
437+
// publishing a Disconnected event on the transition.
435438
func (c *IggyTcpClient) invalidateConnLocked() {
436439
if c.conn != nil {
437440
_ = c.conn.Close()
438441
}
439-
c.state = iggcon.StateDisconnected
442+
if c.state != iggcon.StateDisconnected {
443+
c.state = iggcon.StateDisconnected
444+
c.events.publish(iggcon.DiagnosticEventDisconnected)
445+
}
440446
}
441447

442448
func (c *IggyTcpClient) GetConnectionInfo() *iggcon.ConnectionInfo {
@@ -548,7 +554,7 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error {
548554
if !c.config.reconnection.enabled {
549555
c.logger.Warn("Automatic reconnection is disabled.")
550556
}
551-
// TODO publish event disconnected
557+
c.events.publish(iggcon.DiagnosticEventDisconnected)
552558
return err
553559
}
554560

@@ -558,6 +564,7 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error {
558564
c.connectedAt = time.Now()
559565
c.logger.Info("Iggy client has connected to the Iggy server", slog.String("client_address", c.clientAddress), slog.String("server_address", c.currentServerAddress))
560566
c.mtx.Unlock()
567+
c.events.publish(iggcon.DiagnosticEventConnected)
561568
return nil
562569
}
563570

@@ -624,7 +631,7 @@ func (c *IggyTcpClient) disconnect() error {
624631
}
625632

626633
c.logger.Info("Iggy client has disconnected from server.", slog.String("client_address", c.clientAddress))
627-
// TODO event pushing logic
634+
c.events.publish(iggcon.DiagnosticEventDisconnected)
628635
return nil
629636
}
630637

@@ -646,10 +653,18 @@ func (c *IggyTcpClient) shutdown() error {
646653

647654
c.state = iggcon.StateShutdown
648655
c.logger.Info("Iggy TCP client has been shutdown.", slog.String("client_address", c.clientAddress))
649-
// TODO push shutdown event
656+
c.events.publish(iggcon.DiagnosticEventShutdown)
657+
c.events.close()
650658
return nil
651659
}
652660

653661
func (c *IggyTcpClient) Close() error {
654662
return c.shutdown()
655663
}
664+
665+
// SubscribeEvents returns an independent channel of client lifecycle events
666+
// and an unsubscribe function. The channel is closed on shutdown, after the
667+
// final DiagnosticEventShutdown.
668+
func (c *IggyTcpClient) SubscribeEvents() (<-chan iggcon.DiagnosticEvent, func()) {
669+
return c.events.subscribe()
670+
}

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

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
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+
// subscribe registers a new subscriber and returns its event channel together
46+
// with an unsubscribe function. Calling unsubscribe removes the subscriber and
47+
// closes its channel; it is idempotent and safe to call after the broadcaster
48+
// has been closed.
49+
func (b *eventBroadcaster) subscribe() (<-chan iggcon.DiagnosticEvent, func()) {
50+
b.mu.Lock()
51+
defer b.mu.Unlock()
52+
53+
ch := make(chan iggcon.DiagnosticEvent, subscriberBufferSize)
54+
if b.closed {
55+
close(ch)
56+
return ch, func() {}
57+
}
58+
b.subscribers = append(b.subscribers, ch)
59+
60+
var once sync.Once
61+
unsubscribe := func() {
62+
once.Do(func() {
63+
b.mu.Lock()
64+
defer b.mu.Unlock()
65+
// close() already closed and cleared every channel; nothing to do.
66+
if b.closed {
67+
return
68+
}
69+
for i, sub := range b.subscribers {
70+
if sub == ch {
71+
b.subscribers = append(b.subscribers[:i], b.subscribers[i+1:]...)
72+
close(ch)
73+
return
74+
}
75+
}
76+
})
77+
}
78+
return ch, unsubscribe
79+
}
80+
81+
func (b *eventBroadcaster) publish(event iggcon.DiagnosticEvent) {
82+
b.mu.Lock()
83+
defer b.mu.Unlock()
84+
85+
if b.closed {
86+
return
87+
}
88+
for _, ch := range b.subscribers {
89+
select {
90+
case ch <- event:
91+
default:
92+
// Subscriber is not draining fast enough. Drop the oldest event
93+
// to make room for the newest, preferring recency. Keeps the
94+
// publisher non-blocking even with a slow subscriber.
95+
select {
96+
case <-ch:
97+
default:
98+
}
99+
select {
100+
case ch <- event:
101+
default:
102+
}
103+
}
104+
}
105+
}
106+
107+
func (b *eventBroadcaster) close() {
108+
b.mu.Lock()
109+
defer b.mu.Unlock()
110+
111+
if b.closed {
112+
return
113+
}
114+
b.closed = true
115+
for _, ch := range b.subscribers {
116+
close(ch)
117+
}
118+
b.subscribers = nil
119+
}

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

Lines changed: 151 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,151 @@
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+
}
121+
122+
func TestEventBroadcaster_UnsubscribeStopsDelivery(t *testing.T) {
123+
b := newEventBroadcaster()
124+
ch, unsubscribe := b.subscribe()
125+
126+
unsubscribe()
127+
128+
// Channel must be closed and receive no further events.
129+
if _, ok := <-ch; ok {
130+
t.Fatal("expected channel to be closed after unsubscribe")
131+
}
132+
b.publish(iggcon.DiagnosticEventConnected)
133+
134+
// Remaining subscribers keep receiving.
135+
other, _ := b.subscribe()
136+
b.publish(iggcon.DiagnosticEventSignedIn)
137+
if ev, _ := recvWithTimeout(t, other); ev != iggcon.DiagnosticEventSignedIn {
138+
t.Errorf("remaining subscriber got %v, want signed_in", ev)
139+
}
140+
}
141+
142+
func TestEventBroadcaster_UnsubscribeIsIdempotentAndSafeAfterClose(t *testing.T) {
143+
b := newEventBroadcaster()
144+
_, unsubscribe := b.subscribe()
145+
146+
unsubscribe()
147+
unsubscribe() // second call must not panic or double-close
148+
149+
b.close()
150+
unsubscribe() // after close must not panic
151+
}

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

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ func (c *IggyTcpClient) LoginUser(ctx context.Context, username string, password
5151
}
5252
return c.LoginUser(ctx, username, password)
5353
}
54+
c.events.publish(iggcon.DiagnosticEventSignedIn)
5455
return identity, nil
5556
}
5657

@@ -75,12 +76,16 @@ func (c *IggyTcpClient) LoginWithPersonalAccessToken(ctx context.Context, token
7576
}
7677
return c.LoginWithPersonalAccessToken(ctx, token)
7778
}
79+
c.events.publish(iggcon.DiagnosticEventSignedIn)
7880
return identity, nil
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) {
@@ -117,6 +122,8 @@ func (c *IggyTcpClient) HandleLeaderRedirection(ctx context.Context) (bool, erro
117122
}
118123
c.mtx.Unlock()
119124

125+
c.events.publish(iggcon.DiagnosticEventRedirected)
126+
120127
if err = c.disconnect(); err != nil {
121128
return false, err
122129
}

0 commit comments

Comments
 (0)