167 lines
3.3 KiB
Go
167 lines
3.3 KiB
Go
package client
|
|
|
|
import (
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
type endpoints struct {
|
|
curr int
|
|
list []string
|
|
}
|
|
|
|
func newEndpoints(ee []string) *endpoints {
|
|
return &endpoints{
|
|
curr: 0,
|
|
list: ee,
|
|
}
|
|
}
|
|
|
|
// next returns the next endpoint and its index
|
|
// to try to connect to.
|
|
// Returns -1 index if there is no known RPC endpoints.
|
|
func (e *endpoints) next() (string, int) {
|
|
if len(e.list) == 0 {
|
|
return "", -1
|
|
}
|
|
|
|
next := e.curr + 1
|
|
if next == len(e.list) {
|
|
next = 0
|
|
}
|
|
|
|
e.curr = next
|
|
|
|
return e.list[next], next
|
|
}
|
|
|
|
// current returns an endpoint and its index the Client
|
|
// is connected to.
|
|
// Returns -1 index if there is no known RPC endpoints
|
|
func (e *endpoints) current() (string, int) {
|
|
if len(e.list) == 0 {
|
|
return "", -1
|
|
}
|
|
|
|
return e.list[e.curr], e.curr
|
|
}
|
|
|
|
func (c *Client) switchRPC() bool {
|
|
c.switchLock.Lock()
|
|
defer c.switchLock.Unlock()
|
|
|
|
c.client.Close()
|
|
|
|
_, currEndpointIndex := c.endpoints.current()
|
|
if currEndpointIndex == -1 {
|
|
// there are no known RPC endpoints to try
|
|
// to connect to => do not switch
|
|
return false
|
|
}
|
|
|
|
for {
|
|
newEndpoint, index := c.endpoints.next()
|
|
if index == currEndpointIndex {
|
|
// all the endpoint have been tried
|
|
// for connection unsuccessfully
|
|
return false
|
|
}
|
|
|
|
cli, err := newWSClient(c.cfg, newEndpoint)
|
|
if err != nil {
|
|
c.logger.Warn("could not establish connection to the switched RPC node",
|
|
zap.String("endpoint", newEndpoint),
|
|
zap.Error(err),
|
|
)
|
|
|
|
continue
|
|
}
|
|
|
|
err = cli.Init()
|
|
if err != nil {
|
|
cli.Close()
|
|
c.logger.Warn("could not init the switched RPC node",
|
|
zap.String("endpoint", newEndpoint),
|
|
zap.Error(err),
|
|
)
|
|
|
|
continue
|
|
}
|
|
|
|
c.cache.invalidate()
|
|
c.client = cli
|
|
|
|
return true
|
|
}
|
|
}
|
|
|
|
func (c *Client) notificationLoop() {
|
|
for {
|
|
select {
|
|
case <-c.cfg.ctx.Done():
|
|
_ = c.UnsubscribeAll()
|
|
c.close()
|
|
|
|
return
|
|
case <-c.closeChan:
|
|
_ = c.UnsubscribeAll()
|
|
c.close()
|
|
|
|
return
|
|
case n, ok := <-c.client.Notifications:
|
|
// notification channel is used as a connection
|
|
// state: if it is closed, the connection is
|
|
// considered to be lost
|
|
if !ok {
|
|
var closeReason string
|
|
if closeErr := c.client.GetError(); closeErr != nil {
|
|
closeReason = closeErr.Error()
|
|
} else {
|
|
closeReason = "unknown"
|
|
}
|
|
|
|
c.logger.Warn("switching to the next RPC node",
|
|
zap.String("reason", closeReason),
|
|
)
|
|
|
|
if !c.switchRPC() {
|
|
c.logger.Error("could not establish connection to any RPC node")
|
|
|
|
// could not connect to all endpoints =>
|
|
// switch client to inactive mode
|
|
c.inactiveMode()
|
|
|
|
return
|
|
}
|
|
|
|
newEndpoint, _ := c.endpoints.current()
|
|
|
|
c.logger.Warn("connection to the new RPC node has been established",
|
|
zap.String("endpoint", newEndpoint),
|
|
)
|
|
|
|
if !c.restoreSubscriptions() {
|
|
// new WS client does not allow
|
|
// restoring subscription, client
|
|
// could not work correctly =>
|
|
// closing connection to RPC node
|
|
// to switch to another one
|
|
c.client.Close()
|
|
}
|
|
|
|
// TODO(@carpawell): call here some callback retrieved in constructor
|
|
// of the client to allow checking chain state since during switch
|
|
// process some notification could be lost
|
|
|
|
continue
|
|
}
|
|
|
|
c.notifications <- n
|
|
}
|
|
}
|
|
}
|
|
|
|
// close closes notification channel and wrapped WS client
|
|
func (c *Client) close() {
|
|
close(c.notifications)
|
|
c.client.Close()
|
|
}
|