Skip to content

Commit

Permalink
fix(tcp): recv on new endpoint
Browse files Browse the repository at this point in the history
  • Loading branch information
fumiama committed Jul 16, 2024
1 parent 39d8d5b commit 9a63b3c
Show file tree
Hide file tree
Showing 3 changed files with 39 additions and 3 deletions.
9 changes: 7 additions & 2 deletions gold/link/listen.go
Original file line number Diff line number Diff line change
Expand Up @@ -106,8 +106,13 @@ func (m *Me) dispatch(packet *head.Packet, addr p2p.EndPoint, index int, finish
return
}
if p.endpoint == nil || !p.endpoint.Euqal(addr) {
logrus.Infoln("[listen] @", index, "set endpoint of peer", p.peerip, "to", addr.String())
p.endpoint = addr
if m.ep.Network() == "udp" {
logrus.Infoln("[listen] @", index, "set endpoint of peer", p.peerip, "to", addr.String())
p.endpoint = addr
} else if !addr.Euqal(p.endpoint) && p.endpoint == nil { // tcp/ws, ep not registered
logrus.Infoln("[listen] @", index, "set endpoint of peer", p.peerip, "to", addr.String())
p.endpoint = addr
}
}
switch {
case p.IsToMe(packet.Dst):
Expand Down
30 changes: 29 additions & 1 deletion gold/p2p/tcp/tcp.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,9 @@ func (ep *EndPoint) Network() string {
}

func (ep *EndPoint) Euqal(ep2 p2p.EndPoint) bool {
if ep == nil || ep2 == nil {
return ep == nil && ep2 == nil
}
tcpep2, ok := ep2.(*EndPoint)
if !ok {
return false
Expand Down Expand Up @@ -113,6 +116,10 @@ func (conn *Conn) accept() {
}

func (conn *Conn) receive(ep *EndPoint) {
dialtimeout := conn.addr.dialtimeout
if dialtimeout < time.Second {
dialtimeout = time.Second
}
for {
r := &connrecv{addr: ep}
if conn.addr == nil || conn.lstn == nil || conn.peers == nil || conn.recv == nil {
Expand All @@ -123,7 +130,27 @@ func (conn *Conn) receive(ep *EndPoint) {
return
}
r.conn = tcpconn
_, err := io.Copy(&r.pckt, tcpconn)

stopch := make(chan struct{})
t := time.AfterFunc(dialtimeout, func() {
stopch <- struct{}{}
})

var err error
copych := make(chan struct{})
go func() {
_, err = io.Copy(&r.pckt, tcpconn)
copych <- struct{}{}
}()

select {
case <-stopch:
logrus.Debugln("[tcp] recv from", ep, "timeout")
return
case <-copych:
t.Stop()
}

if err != nil {
logrus.Debugln("[tcp] recv from", ep, "err:", err)
return
Expand Down Expand Up @@ -211,6 +238,7 @@ func (conn *Conn) WriteToPeer(b []byte, ep p2p.EndPoint) (n int, err error) {
})
logrus.Debugln("[tcp] dial to", tcpep.addr, "success, local:", tcpconn.LocalAddr())
conn.peers.Set(tcpep.String(), tcpconn)
go conn.receive(tcpep)
} else {
logrus.Debugln("[tcp] reuse tcpconn from", tcpconn.LocalAddr(), "to", tcpconn.RemoteAddr())
}
Expand Down
3 changes: 3 additions & 0 deletions gold/p2p/udp/udp.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@ func (ep *EndPoint) Network() string {
}

func (ep *EndPoint) Euqal(ep2 p2p.EndPoint) bool {
if ep == nil || ep2 == nil {
return ep == nil && ep2 == nil
}
udpep2, ok := ep2.(*EndPoint)
if !ok {
return false
Expand Down

0 comments on commit 9a63b3c

Please sign in to comment.