Code cleaning

This commit is contained in:
Zoltan Papp 2024-05-28 01:27:53 +02:00
parent 076ce69a24
commit 6d627f1923

View File

@ -39,15 +39,15 @@ type Client struct {
serverAddress string serverAddress string
hashedID []byte hashedID []byte
relayConnIsEstablished bool readyToOpenConns bool
conns map[string]*connContainer conns map[string]*connContainer
connsMutext sync.Mutex // protect conns and relayConnIsEstablished bool connsMutext sync.Mutex // protect conns and readyToOpenConns bool
relayConn net.Conn relayConn net.Conn
serviceIsRunning bool serviceIsRunning bool
wgRelayConn sync.WaitGroup serviceIsRunningMutex sync.Mutex
mu sync.Mutex wgReadLoop sync.WaitGroup
onDisconnected chan struct{} onDisconnected chan struct{}
} }
func NewClient(ctx context.Context, serverAddress, peerID string) *Client { func NewClient(ctx context.Context, serverAddress, peerID string) *Client {
@ -65,24 +65,24 @@ func NewClient(ctx context.Context, serverAddress, peerID string) *Client {
} }
func (c *Client) Connect() error { func (c *Client) Connect() error {
c.mu.Lock() c.serviceIsRunningMutex.Lock()
if c.serviceIsRunning { if c.serviceIsRunning {
c.mu.Unlock() c.serviceIsRunningMutex.Unlock()
return nil return nil
} }
err := c.connect() err := c.connect()
if err != nil { if err != nil {
c.mu.Unlock() c.serviceIsRunningMutex.Unlock()
return err return err
} }
c.serviceIsRunning = true c.serviceIsRunning = true
c.wgRelayConn.Add(1) c.wgReadLoop.Add(1)
go c.readLoop() go c.readLoop()
c.mu.Unlock() c.serviceIsRunningMutex.Unlock()
go func() { go func() {
<-c.ctx.Done() <-c.ctx.Done()
@ -99,11 +99,11 @@ func (c *Client) Connect() error {
func (c *Client) reconnectGuard() { func (c *Client) reconnectGuard() {
for { for {
c.wgRelayConn.Wait() c.wgReadLoop.Wait()
c.mu.Lock() c.serviceIsRunningMutex.Lock()
if !c.serviceIsRunning { if !c.serviceIsRunning {
c.mu.Unlock() c.serviceIsRunningMutex.Unlock()
return return
} }
@ -111,31 +111,24 @@ func (c *Client) reconnectGuard() {
err := c.connect() err := c.connect()
if err != nil { if err != nil {
log.Errorf("failed to reconnect to relay server: %s", err) log.Errorf("failed to reconnect to relay server: %s", err)
c.mu.Unlock() c.serviceIsRunningMutex.Unlock()
time.Sleep(reconnectingTimeout) time.Sleep(reconnectingTimeout)
continue continue
} }
log.Infof("reconnected to relay server") log.Infof("reconnected to relay server")
c.wgRelayConn.Add(1) c.wgReadLoop.Add(1)
go c.readLoop() go c.readLoop()
c.mu.Unlock() c.serviceIsRunningMutex.Unlock()
} }
} }
func (c *Client) OpenConn(dstPeerID string) (net.Conn, error) { func (c *Client) OpenConn(dstPeerID string) (net.Conn, error) {
c.mu.Lock()
defer c.mu.Unlock()
c.connsMutext.Lock() c.connsMutext.Lock()
defer c.connsMutext.Unlock() defer c.connsMutext.Unlock()
if !c.relayConnIsEstablished { if !c.readyToOpenConns {
return nil, fmt.Errorf("relay connection is not established")
}
if !c.serviceIsRunning {
return nil, fmt.Errorf("relay connection is not established") return nil, fmt.Errorf("relay connection is not established")
} }
@ -152,6 +145,12 @@ func (c *Client) OpenConn(dstPeerID string) (net.Conn, error) {
} }
func (c *Client) Close() error { func (c *Client) Close() error {
c.serviceIsRunningMutex.Lock()
if !c.serviceIsRunning {
c.serviceIsRunningMutex.Unlock()
return nil
}
c.ctxCancel() c.ctxCancel()
return c.close() return c.close()
} }
@ -173,13 +172,13 @@ func (c *Client) connect() error {
return err return err
} }
c.relayConnIsEstablished = true c.readyToOpenConns = true
return nil return nil
} }
func (c *Client) close() error { func (c *Client) close() error {
c.mu.Lock() c.serviceIsRunningMutex.Lock()
defer c.mu.Unlock() defer c.serviceIsRunningMutex.Unlock()
if !c.serviceIsRunning { if !c.serviceIsRunning {
return nil return nil
@ -189,7 +188,7 @@ func (c *Client) close() error {
err := c.relayConn.Close() err := c.relayConn.Close()
c.wgRelayConn.Wait() c.wgReadLoop.Wait()
return err return err
} }
@ -284,7 +283,7 @@ func (c *Client) readLoop() {
} }
c.connsMutext.Lock() c.connsMutext.Lock()
c.relayConnIsEstablished = false c.readyToOpenConns = false
for _, container := range c.conns { for _, container := range c.conns {
close(container.messages) close(container.messages)
} }
@ -292,17 +291,17 @@ func (c *Client) readLoop() {
c.connsMutext.Unlock() c.connsMutext.Unlock()
c.log.Tracef("exit from read loop") c.log.Tracef("exit from read loop")
c.wgRelayConn.Done() c.wgReadLoop.Done()
} }
func (c *Client) writeTo(id string, dstID []byte, payload []byte) (int, error) { func (c *Client) writeTo(id string, dstID []byte, payload []byte) (int, error) {
c.mu.Lock() c.connsMutext.Lock()
_, ok := c.conns[id] _, ok := c.conns[id]
if !ok { if !ok {
c.mu.Unlock() c.connsMutext.Unlock()
return 0, io.EOF return 0, io.EOF
} }
c.mu.Unlock() c.connsMutext.Unlock()
msg := messages.MarshalTransportMsg(dstID, payload) msg := messages.MarshalTransportMsg(dstID, payload)
n, err := c.relayConn.Write(msg) n, err := c.relayConn.Write(msg)
if err != nil { if err != nil {
@ -329,9 +328,6 @@ func (c *Client) generateConnReaderFN(msgChannel chan Msg) func(b []byte) (n int
} }
func (c *Client) closeConn(id string) error { func (c *Client) closeConn(id string) error {
c.mu.Lock()
defer c.mu.Unlock()
c.connsMutext.Lock() c.connsMutext.Lock()
defer c.connsMutext.Unlock() defer c.connsMutext.Unlock()