forked from hazelcast/hazelcast-go-client
/
view_listener_service.go
68 lines (60 loc) · 2.11 KB
/
view_listener_service.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
package cluster
import (
"context"
"fmt"
"sync/atomic"
"github.com/JorgenPo/hazelcast-go-client/internal/event"
"github.com/JorgenPo/hazelcast-go-client/internal/logger"
)
type ViewListenerService struct {
cs *Service
cm *ConnectionManager
dispatcher *event.DispatchService
logger logger.Logger
connID int64
}
func NewViewListenerService(cs *Service, cm *ConnectionManager, dispatcher *event.DispatchService, logger logger.Logger) *ViewListenerService {
vs := &ViewListenerService{
cs: cs,
cm: cm,
dispatcher: dispatcher,
logger: logger,
}
dispatcher.Subscribe(EventConnectionOpened, event.DefaultSubscriptionID, vs.handleConnectionOpened)
dispatcher.Subscribe(EventConnectionClosed, event.DefaultSubscriptionID, vs.handleConnectionClosed)
return vs
}
func (vs *ViewListenerService) handleConnectionOpened(event event.Event) {
vs.logger.Trace(func() string { return "cluster.ViewListenerService.handleConnectionOpened" })
if e, ok := event.(*ConnectionOpened); ok {
vs.tryRegister(e.Conn)
}
}
func (vs *ViewListenerService) handleConnectionClosed(event event.Event) {
vs.logger.Trace(func() string { return "cluster.ViewListenerService.handleConnectionClosed" })
if e, ok := event.(*ConnectionClosed); ok {
vs.tryReregisterToRandomConnection(e.Conn)
}
}
func (vs *ViewListenerService) tryRegister(conn *Connection) {
vs.logger.Trace(func() string {
return fmt.Sprintf("cluster.ViewListenerService.tryRegister (status: %d): %d", atomic.LoadInt32(&conn.status), conn.connectionID)
})
if !atomic.CompareAndSwapInt64(&vs.connID, 0, conn.connectionID) {
return
}
if err := vs.cs.sendMemberListViewRequest(context.Background(), conn); err != nil {
vs.tryReregisterToRandomConnection(conn)
}
}
func (vs *ViewListenerService) tryReregisterToRandomConnection(oldConn *Connection) {
vs.logger.Trace(func() string {
return fmt.Sprintf("cluster.ViewListenerService.tryReRegister: %d", oldConn.connectionID)
})
if !atomic.CompareAndSwapInt64(&vs.connID, oldConn.connectionID, 0) {
return
}
if conn := vs.cm.RandomConnection(); conn != nil {
vs.tryRegister(conn)
}
}