Merge pull request #16 from anyproto/peer-active-subconn-gc
peer: gc active sub conn
This commit is contained in:
commit
fb1df54941
@ -222,8 +222,15 @@ func (p *peer) gc(ttl time.Duration) {
|
||||
select {
|
||||
case <-act.Closed():
|
||||
delete(p.active, act)
|
||||
continue
|
||||
default:
|
||||
}
|
||||
if act.LastUsage().Before(minLastUsage) {
|
||||
log.Warn("close active connection because no activity", zap.String("peerId", p.id), zap.String("addr", p.Addr()))
|
||||
_ = act.Close()
|
||||
delete(p.active, act)
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -105,6 +105,7 @@ func TestPeer_TryClose(t *testing.T) {
|
||||
}()
|
||||
defer out2.Close()
|
||||
fx.mc.EXPECT().Open(gomock.Any()).Return(in2, nil)
|
||||
fx.mc.EXPECT().Addr().Return("")
|
||||
dc2, err := fx.AcquireDrpcConn(ctx)
|
||||
require.NoError(t, err)
|
||||
_ = dc2.Close()
|
||||
@ -118,6 +119,18 @@ func TestPeer_TryClose(t *testing.T) {
|
||||
fx.mc.EXPECT().Open(gomock.Any()).Return(in3, nil)
|
||||
dc3, err := fx.AcquireDrpcConn(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
// make another active, should be removed by ttl
|
||||
in4, out4 := net.Pipe()
|
||||
go func() {
|
||||
handshake.IncomingProtoHandshake(ctx, out4, defaultProtoChecker)
|
||||
}()
|
||||
defer out4.Close()
|
||||
fx.mc.EXPECT().Open(gomock.Any()).Return(in4, nil)
|
||||
dc4, err := fx.AcquireDrpcConn(ctx)
|
||||
require.NoError(t, err)
|
||||
defer dc4.Close()
|
||||
|
||||
fx.ReleaseDrpcConn(dc3)
|
||||
_ = dc3.Close()
|
||||
fx.ReleaseDrpcConn(dc)
|
||||
|
||||
@ -53,7 +53,9 @@ type secureService struct {
|
||||
}
|
||||
|
||||
func (s *secureService) Init(a *app.App) (err error) {
|
||||
s.protoVersion = ProtoVersion
|
||||
if s.protoVersion == 0 {
|
||||
s.protoVersion = ProtoVersion
|
||||
}
|
||||
account := a.MustComponent(commonaccount.CName).(commonaccount.Service)
|
||||
peerKey, err := account.Account().PeerKey.Raw()
|
||||
if err != nil {
|
||||
|
||||
@ -57,7 +57,7 @@ func TestHandshake(t *testing.T) {
|
||||
|
||||
func TestHandshakeIncompatibleVersion(t *testing.T) {
|
||||
nc := testnodeconf.GenNodeConfig(2)
|
||||
fxS := newFixture(t, nc, nc.GetAccountService(0), 0)
|
||||
fxS := newFixture(t, nc, nc.GetAccountService(0), 1)
|
||||
defer fxS.Finish(t)
|
||||
sc, cc := net.Pipe()
|
||||
|
||||
@ -72,7 +72,7 @@ func TestHandshakeIncompatibleVersion(t *testing.T) {
|
||||
ar.ctx, ar.err = fxS.SecureInbound(ctx, sc)
|
||||
resCh <- ar
|
||||
}()
|
||||
fxC := newFixture(t, nc, nc.GetAccountService(1), 1)
|
||||
fxC := newFixture(t, nc, nc.GetAccountService(1), 2)
|
||||
defer fxC.Finish(t)
|
||||
_, err := fxC.SecureOutbound(ctx, cc)
|
||||
require.Equal(t, handshake.ErrIncompatibleVersion, err)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user