Skip to content

Commit e73e0a5

Browse files
kylecarbskylecarbs
committed
Merge branch 'main' into agentroutes
2 parents e903072 + 55784d3 commit e73e0a5

File tree

14 files changed

+133
-521
lines changed

14 files changed

+133
-521
lines changed

‎agent/agent_test.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -95,7 +95,7 @@ func TestAgent(t *testing.T) {
9595
func setup(t *testing.T) proto.DRPCPeerBrokerClient {
9696
client, server := provisionersdk.TransportPipe()
9797
closer := agent.New(func(ctx context.Context, opts *peer.ConnOptions) (*peerbroker.Listener, error) {
98-
return peerbroker.Listen(server, opts)
98+
return peerbroker.Listen(server, nil, opts)
9999
}, &peer.ConnOptions{
100100
Logger: slogtest.Make(t, nil).Leveled(slog.LevelDebug),
101101
})

‎cli/projectcreate.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -118,7 +118,7 @@ func projectCreate() *cobra.Command {
118118
return cmd
119119
}
120120

121-
func validateProjectVersionSource(cmd *cobra.Command, client *codersdk.Client, organization coderd.Organization, provisioner database.ProvisionerType, directory string, parameters ...coderd.CreateParameterValueRequest) (*coderd.ProjectVersion, error) {
121+
func validateProjectVersionSource(cmd *cobra.Command, client *codersdk.Client, organization coderd.Organization, provisioner database.ProvisionerType, directory string, parameters ...coderd.CreateParameterRequest) (*coderd.ProjectVersion, error) {
122122
spin := spinner.New(spinner.CharSets[5], 100*time.Millisecond)
123123
spin.Writer = cmd.OutOrStdout()
124124
spin.Suffix = " Uploading current directory..."
@@ -192,7 +192,7 @@ func validateProjectVersionSource(cmd *cobra.Command, client *codersdk.Client, o
192192
if err != nil {
193193
return nil, err
194194
}
195-
parameters = append(parameters, coderd.CreateParameterValueRequest{
195+
parameters = append(parameters, coderd.CreateParameterRequest{
196196
Name: parameterSchema.Name,
197197
SourceValue: value,
198198
SourceScheme: database.ParameterSourceSchemeData,

‎codersdk/workspaceresources.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -106,5 +106,5 @@ func (c *Client) ListenWorkspaceAgent(ctx context.Context, opts *peer.ConnOption
106106
if err != nil {
107107
return nil, xerrors.Errorf("multiplex client: %w", err)
108108
}
109-
return peerbroker.Listen(session, opts)
109+
return peerbroker.Listen(session, nil, opts)
110110
}

‎peerbroker/dial.go

Lines changed: 8 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -12,36 +12,6 @@ import (
1212

1313
// Dial consumes the PeerBroker gRPC connection negotiation stream to produce a WebRTC peered connection.
1414
func Dial(stream proto.DRPCPeerBroker_NegotiateConnectionClient, iceServers []webrtc.ICEServer, opts *peer.ConnOptions) (*peer.Conn, error) {
15-
// Convert WebRTC ICE servers to the protobuf type.
16-
protoIceServers := make([]*proto.WebRTCICEServer, 0, len(iceServers))
17-
for _, iceServer := range iceServers {
18-
var credentialString string
19-
if value, ok := iceServer.Credential.(string); ok {
20-
credentialString = value
21-
}
22-
protoIceServers = append(protoIceServers, &proto.WebRTCICEServer{
23-
Urls: iceServer.URLs,
24-
Username: iceServer.Username,
25-
Credential: credentialString,
26-
CredentialType: int32(iceServer.CredentialType),
27-
})
28-
}
29-
if len(protoIceServers) > 0 {
30-
// Send ICE servers to connect with.
31-
// Client sends ICE servers so clients can determine the node
32-
// servers will meet at. eg. us-west1.coder.com could be a TURN server.
33-
err := stream.Send(&proto.NegotiateConnection_ClientToServer{
34-
Message: &proto.NegotiateConnection_ClientToServer_Servers{
35-
Servers: &proto.WebRTCICEServers{
36-
Servers: protoIceServers,
37-
},
38-
},
39-
})
40-
if err != nil {
41-
return nil, xerrors.Errorf("write ice servers: %w", err)
42-
}
43-
}
44-
4515
peerConn, err := peer.Client(iceServers, opts)
4616
if err != nil {
4717
return nil, xerrors.Errorf("create peer connection: %w", err)
@@ -54,9 +24,9 @@ func Dial(stream proto.DRPCPeerBroker_NegotiateConnectionClient, iceServers []we
5424
case <-peerConn.Closed():
5525
return
5626
case sessionDescription := <-peerConn.LocalSessionDescription():
57-
err = stream.Send(&proto.NegotiateConnection_ClientToServer{
58-
Message: &proto.NegotiateConnection_ClientToServer_Offer{
59-
Offer: &proto.WebRTCSessionDescription{
27+
err = stream.Send(&proto.Exchange{
28+
Message: &proto.Exchange_Sdp{
29+
Sdp: &proto.WebRTCSessionDescription{
6030
SdpType: int32(sessionDescription.Type),
6131
Sdp: sessionDescription.SDP,
6232
},
@@ -67,8 +37,8 @@ func Dial(stream proto.DRPCPeerBroker_NegotiateConnectionClient, iceServers []we
6737
return
6838
}
6939
case iceCandidate := <-peerConn.LocalCandidate():
70-
err = stream.Send(&proto.NegotiateConnection_ClientToServer{
71-
Message: &proto.NegotiateConnection_ClientToServer_IceCandidate{
40+
err = stream.Send(&proto.Exchange{
41+
Message: &proto.Exchange_IceCandidate{
7242
IceCandidate: iceCandidate.Candidate,
7343
},
7444
})
@@ -89,10 +59,10 @@ func Dial(stream proto.DRPCPeerBroker_NegotiateConnectionClient, iceServers []we
8959
}
9060

9161
switch {
92-
case serverToClientMessage.GetAnswer() != nil:
62+
case serverToClientMessage.GetSdp() != nil:
9363
peerConn.SetRemoteSessionDescription(webrtc.SessionDescription{
94-
Type: webrtc.SDPType(serverToClientMessage.GetAnswer().SdpType),
95-
SDP: serverToClientMessage.GetAnswer().Sdp,
64+
Type: webrtc.SDPType(serverToClientMessage.GetSdp().SdpType),
65+
SDP: serverToClientMessage.GetSdp().Sdp,
9666
})
9767
case serverToClientMessage.GetIceCandidate() != "":
9868
peerConn.AddRemoteCandidate(webrtc.ICECandidateInit{

‎peerbroker/dial_test.go

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -31,18 +31,26 @@ func TestDial(t *testing.T) {
3131
defer client.Close()
3232
defer server.Close()
3333

34-
listener, err := peerbroker.Listen(server, &peer.ConnOptions{
35-
Logger: slogtest.Make(t, nil).Named("server").Leveled(slog.LevelDebug),
34+
settingEngine := webrtc.SettingEngine{}
35+
listener, err := peerbroker.Listen(server, func(ctx context.Context) ([]webrtc.ICEServer, error) {
36+
return []webrtc.ICEServer{{
37+
URLs: []string{"stun:stun.l.google.com:19302"},
38+
}}, nil
39+
}, &peer.ConnOptions{
40+
Logger: slogtest.Make(t, nil).Named("server").Leveled(slog.LevelDebug),
41+
SettingEngine: settingEngine,
3642
})
3743
require.NoError(t, err)
3844

3945
api := proto.NewDRPCPeerBrokerClient(provisionersdk.Conn(client))
4046
stream, err := api.NegotiateConnection(ctx)
4147
require.NoError(t, err)
48+
4249
clientConn, err := peerbroker.Dial(stream, []webrtc.ICEServer{{
4350
URLs: []string{"stun:stun.l.google.com:19302"},
4451
}}, &peer.ConnOptions{
45-
Logger: slogtest.Make(t, nil).Named("client").Leveled(slog.LevelDebug),
52+
Logger: slogtest.Make(t, nil).Named("client").Leveled(slog.LevelDebug),
53+
SettingEngine: settingEngine,
4654
})
4755
require.NoError(t, err)
4856
defer clientConn.Close()

‎peerbroker/listen.go

Lines changed: 24 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -17,13 +17,22 @@ import (
1717
"github.com/coder/coder/peerbroker/proto"
1818
)
1919

20+
// ICEServersFunc returns ICEServers when a new connection is requested.
21+
type ICEServersFunc func(ctx context.Context) ([]webrtc.ICEServer, error)
22+
2023
// Listen consumes the transport as the server-side of the PeerBroker dRPC service.
2124
// The Accept function must be serviced, or new connections will hang.
22-
func Listen(connListener net.Listener, opts *peer.ConnOptions) (*Listener, error) {
25+
func Listen(connListener net.Listener, iceServersFunc ICEServersFunc, opts *peer.ConnOptions) (*Listener, error) {
26+
if iceServersFunc == nil {
27+
iceServersFunc = func(ctx context.Context) ([]webrtc.ICEServer, error) {
28+
return []webrtc.ICEServer{}, nil
29+
}
30+
}
2331
ctx, cancelFunc := context.WithCancel(context.Background())
2432
listener := &Listener{
2533
connectionChannel: make(chan *peer.Conn),
2634
connectionListener: connListener,
35+
iceServersFunc: iceServersFunc,
2736

2837
closeFunc: cancelFunc,
2938
closed: make(chan struct{}),
@@ -50,6 +59,7 @@ func Listen(connListener net.Listener, opts *peer.ConnOptions) (*Listener, error
5059
type Listener struct {
5160
connectionChannel chan *peer.Conn
5261
connectionListener net.Listener
62+
iceServersFunc ICEServersFunc
5363

5464
closeFunc context.CancelFunc
5565
closed chan struct{}
@@ -107,8 +117,12 @@ type peerBrokerService struct {
107117

108118
// NegotiateConnection negotiates a WebRTC connection.
109119
func (b *peerBrokerService) NegotiateConnection(stream proto.DRPCPeerBroker_NegotiateConnectionStream) error {
120+
iceServers, err := b.listener.iceServersFunc(stream.Context())
121+
if err != nil {
122+
return xerrors.Errorf("get ice servers: %w", err)
123+
}
110124
// Start with no ICE servers. They can be sent by the client if provided.
111-
peerConn, err := peer.Server([]webrtc.ICEServer{}, b.connOptions)
125+
peerConn, err := peer.Server(iceServers, b.connOptions)
112126
if err != nil {
113127
return xerrors.Errorf("create peer connection: %w", err)
114128
}
@@ -124,9 +138,9 @@ func (b *peerBrokerService) NegotiateConnection(stream proto.DRPCPeerBroker_Nego
124138
case <-peerConn.Closed():
125139
return
126140
case sessionDescription := <-peerConn.LocalSessionDescription():
127-
err = stream.Send(&proto.NegotiateConnection_ServerToClient{
128-
Message: &proto.NegotiateConnection_ServerToClient_Answer{
129-
Answer: &proto.WebRTCSessionDescription{
141+
err = stream.Send(&proto.Exchange{
142+
Message: &proto.Exchange_Sdp{
143+
Sdp: &proto.WebRTCSessionDescription{
130144
SdpType: int32(sessionDescription.Type),
131145
Sdp: sessionDescription.SDP,
132146
},
@@ -137,8 +151,8 @@ func (b *peerBrokerService) NegotiateConnection(stream proto.DRPCPeerBroker_Nego
137151
return
138152
}
139153
case iceCandidate := <-peerConn.LocalCandidate():
140-
err = stream.Send(&proto.NegotiateConnection_ServerToClient{
141-
Message: &proto.NegotiateConnection_ServerToClient_IceCandidate{
154+
err = stream.Send(&proto.Exchange{
155+
Message: &proto.Exchange_IceCandidate{
142156
IceCandidate: iceCandidate.Candidate,
143157
},
144158
})
@@ -159,28 +173,11 @@ func (b *peerBrokerService) NegotiateConnection(stream proto.DRPCPeerBroker_Nego
159173
}
160174

161175
switch {
162-
case clientToServerMessage.GetOffer() != nil:
176+
case clientToServerMessage.GetSdp() != nil:
163177
peerConn.SetRemoteSessionDescription(webrtc.SessionDescription{
164-
Type: webrtc.SDPType(clientToServerMessage.GetOffer().SdpType),
165-
SDP: clientToServerMessage.GetOffer().Sdp,
178+
Type: webrtc.SDPType(clientToServerMessage.GetSdp().SdpType),
179+
SDP: clientToServerMessage.GetSdp().Sdp,
166180
})
167-
case clientToServerMessage.GetServers() != nil:
168-
// Convert protobuf ICE servers to the WebRTC type.
169-
iceServers := make([]webrtc.ICEServer, 0, len(clientToServerMessage.GetServers().Servers))
170-
for _, iceServer := range clientToServerMessage.GetServers().Servers {
171-
iceServers = append(iceServers, webrtc.ICEServer{
172-
URLs: iceServer.Urls,
173-
Username: iceServer.Username,
174-
Credential: iceServer.Credential,
175-
CredentialType: webrtc.ICECredentialType(iceServer.CredentialType),
176-
})
177-
}
178-
err = peerConn.SetConfiguration(webrtc.Configuration{
179-
ICEServers: iceServers,
180-
})
181-
if err != nil {
182-
return peerConn.CloseWithError(xerrors.Errorf("set ice configuration: %w", err))
183-
}
184181
case clientToServerMessage.GetIceCandidate() != "":
185182
peerConn.AddRemoteCandidate(webrtc.ICECandidateInit{
186183
Candidate: clientToServerMessage.GetIceCandidate(),

‎peerbroker/listen_test.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ func TestListen(t *testing.T) {
2323
defer client.Close()
2424
defer server.Close()
2525

26-
listener, err := peerbroker.Listen(server, nil)
26+
listener, err := peerbroker.Listen(server, nil, nil)
2727
require.NoError(t, err)
2828

2929
api := proto.NewDRPCPeerBrokerClient(provisionersdk.Conn(client))
@@ -43,7 +43,7 @@ func TestListen(t *testing.T) {
4343
defer client.Close()
4444
defer server.Close()
4545

46-
listener, err := peerbroker.Listen(server, nil)
46+
listener, err := peerbroker.Listen(server, nil, nil)
4747
require.NoError(t, err)
4848
go listener.Close()
4949
_, err = listener.Accept()

0 commit comments

Comments
 (0)