Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings
This repository was archived by the owner on Aug 30, 2024. It is now read-only.
/coder-v1-cliPublic archive

fix: Only close connection scoped data channels on RTC close#403

Merged
deansheather merged 4 commits intomasterfromconnclose
Jul 27, 2021
Merged
Show file tree
Hide file tree
Changes fromall commits
Commits
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 36 additions & 20 deletionswsnet/dial_test.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -293,39 +293,55 @@ func TestDial(t *testing.T) {

t.Run("Close Listeners on Disconnect", func(t *testing.T) {
t.Parallel()
log := slogtest.Make(t, nil)

tcpListener, err := net.Listen("tcp", "0.0.0.0:0")
listener, err := net.Listen("tcp", "0.0.0.0:0")
require.NoError(t, err)
go func() {
_, _ = tcpListener.Accept()
for {
c, _ := listener.Accept()

go func() {
b := make([]byte, 5)
_, err := c.Read(b)
if err != nil {
return
}
_, err = c.Write(b)
require.NoError(t, err)
}()
}
}()

connectAddr, listenAddr := createDumbBroker(t)
l, err:= Listen(context.Background(), slogtest.Make(t, nil), listenAddr, "")
_, err = Listen(context.Background(), slogtest.Make(t, nil), listenAddr, "")
require.NoError(t, err)

turnAddr, closeTurn := createTURNServer(t, ice.SchemeTypeTURN)
dialer, err := DialWebsocket(context.Background(), connectAddr, &DialOptions{
ICEServers: []webrtc.ICEServer{{
URLs: []string{fmt.Sprintf("turn:%s", turnAddr)},
Username: "example",
Credential: testPass,
CredentialType: webrtc.ICECredentialTypePassword,
}},
d1, err := DialWebsocket(context.Background(), connectAddr, &DialOptions{
Log: &log,
}, nil)
require.NoError(t, err)
_, err = d1.DialContext(context.Background(), listener.Addr().Network(), listener.Addr().String())
require.NoError(t, err)

_, err = dialer.DialContext(context.Background(), "tcp", tcpListener.Addr().String())
d2, err := DialWebsocket(context.Background(), connectAddr, &DialOptions{
Log: &log,
}, nil)
require.NoError(t, err)
conn, err := d2.DialContext(context.Background(), listener.Addr().Network(), listener.Addr().String())
require.NoError(t, err)
err = d1.Close()
require.NoError(t, err)

closeTurn()
// TODO: This needs to be longer than the KeepAlive timeout for the RTC connection.
// Once the listener stores RTC connections instead of io.Closer we can directly
// reference the RTC connection to ensure it's properly closed.
time.Sleep(time.Second * 10)

list := l.(*listener)
assert.Eventually(t, func() bool {
list.connClosersMut.Lock()
defer list.connClosersMut.Unlock()
return len(list.connClosers) == 0
}, time.Second*15, time.Millisecond*100)
b := []byte("hello")
_, err = conn.Write(b)
require.NoError(t, err)
_, err = conn.Read(b)
require.NoError(t, err)
})
}

Expand Down
29 changes: 17 additions & 12 deletionswsnet/listen.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -163,9 +163,11 @@ func (l *listener) negotiate(ctx context.Context, conn net.Conn) {
ctx = slog.With(ctx, slog.F("conn_id", id))

var (
err error
decoder = json.NewDecoder(conn)
rtc *webrtc.PeerConnection
err error
decoder = json.NewDecoder(conn)
rtc *webrtc.PeerConnection
connClosers = make([]io.Closer, 0)
connClosersMut sync.Mutex
// If candidates are sent before an offer, we place them here.
// We currently have no assurances to ensure this can't happen,
// so it's better to buffer and process than fail.
Expand DownExpand Up@@ -255,6 +257,9 @@ func (l *listener) negotiate(ctx context.Context, conn net.Conn) {
closeError(err)
return
}
l.connClosersMut.Lock()
l.connClosers = append(l.connClosers, rtc)
l.connClosersMut.Unlock()
rtc.OnConnectionStateChange(func(pcs webrtc.PeerConnectionState) {
l.log.Info(ctx, "connection state change", slog.F("state", pcs.String()))
switch pcs {
Expand All@@ -267,16 +272,16 @@ func (l *listener) negotiate(ctx context.Context, conn net.Conn) {
}

// Close connections opened when RTC was alive.
l.connClosersMut.Lock()
deferl.connClosersMut.Unlock()
for _, connCloser := rangel.connClosers {
connClosersMut.Lock()
defer connClosersMut.Unlock()
for _, connCloser := range connClosers {
_ = connCloser.Close()
}
l.connClosers = make([]io.Closer, 0)
connClosers = make([]io.Closer, 0)
})

flushCandidates := proxyICECandidates(rtc, conn)
rtc.OnDataChannel(l.handle(ctx, msg))
rtc.OnDataChannel(l.handle(ctx, msg, &connClosers, &connClosersMut))

l.log.Debug(ctx, "set remote description", slog.F("offer", *msg.Offer))
err = rtc.SetRemoteDescription(*msg.Offer)
Expand DownExpand Up@@ -329,7 +334,7 @@ func (l *listener) negotiate(ctx context.Context, conn net.Conn) {
}

// nolint:gocognit
func (l *listener) handle(ctx context.Context, msg BrokerMessage) func(dc *webrtc.DataChannel) {
func (l *listener) handle(ctx context.Context, msg BrokerMessage, connClosers *[]io.Closer, connClosersMut *sync.Mutex) func(dc *webrtc.DataChannel) {
return func(dc *webrtc.DataChannel) {
if dc.Protocol() == controlChannel {
// The control channel handles pings.
Expand DownExpand Up@@ -430,9 +435,9 @@ func (l *listener) handle(ctx context.Context, msg BrokerMessage) func(dc *webrt
dc: dc,
rw: rw,
}
l.connClosersMut.Lock()
l.connClosers = append(l.connClosers, co)
l.connClosersMut.Unlock()
connClosersMut.Lock()
*connClosers = append(*connClosers, co)
connClosersMut.Unlock()
co.init()
defer nc.Close()
defer co.Close()
Expand Down
[8]ページ先頭
©2009-2025 Movatter.jp