Support setting bandwidth for proxy publishers.

cmd/proxy/proxy_server.go

@@ -1231,6 +1231,37 @@ func (s *ProxyServer) processCommand(ctx context.Context, client *ProxyClient, s
 			client.Close(context.Background())
 		}()
 
+		response := &proxy.ServerMessage{
+			Id:   message.Id,
+			Type: "command",
+			Command: &proxy.CommandServerMessage{
+				Id: cmd.ClientId,
+			},
+		}
+		session.sendMessage(response)
+	case "update-bandwidth":
+		client := s.GetClient(cmd.ClientId)
+		if client == nil {
+			session.sendMessage(message.NewErrorServerMessage(UnknownClient))
+			return
+		}
+
+		clientWithBandwidth, ok := client.(sfu.ClientWithBandwidth)
+		if !ok {
+			session.sendMessage(message.NewErrorServerMessage(UnknownClient))
+			return
+		}
+
+		if cmd.Bandwidth != 0 {
+			ctx2, cancel := context.WithTimeout(ctx, s.mcuTimeout)
+			defer cancel()
+
+			if err := clientWithBandwidth.SetBandwidth(ctx2, cmd.Bandwidth); err != nil {
+				session.sendMessage(message.NewWrappedErrorServerMessage(err))
+				return
+			}
+		}
+
 		response := &proxy.ServerMessage{
 			Id:   message.Id,
 			Type: "command",

proxy/api.go

@@ -219,6 +219,9 @@ type CommandClientMessage struct {
 	Hostname string `json:"hostname,omitempty"`
 	Port     int    `json:"port,omitempty"`
 	RtcpPort int    `json:"rtcpPort,omitempty"`
+
+	// Bandwidth is set if Type is "update-bandwidth"
+	Bandwidth api.Bandwidth `json:"bandwidth,omitempty"`
 }
 
 func (m *CommandClientMessage) CheckValid() error {

sfu/proxy/proxy.go

@@ -139,6 +139,22 @@ func (c *proxyPubSubCommon) Bandwidth() *sfu.ClientBandwidthInfo {
 	return c.bandwidth.Load()
 }
 
+func (c *proxyPubSubCommon) SetBandwidth(ctx context.Context, bandwidth api.Bandwidth) error {
+	_, _, err := c.conn.performSyncRequest(ctx, &proxy.ClientMessage{
+		Type: "command",
+		Command: &proxy.CommandClientMessage{
+			ClientId:  c.proxyId,
+			Type:      "update-bandwidth",
+			Bandwidth: bandwidth,
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func (c *proxyPubSubCommon) doSendMessage(ctx context.Context, msg *proxy.ClientMessage, callback func(error, api.StringMap)) {
 	c.conn.performAsyncRequest(ctx, msg, func(err error, response *proxy.ServerMessage) {
 		if err != nil {