devbox
小蓝xlanll
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

使用ION-SFU和媒体设备在Golang中构建一个WebRTC视频和音频广播器

作者：小蓝xlanll | 2024-02-28 03:46:48

赞

踩

ion-sfu

在本教程中，您将构建一个视频广播应用程序，该应用程序在 Golang 中读取摄像头并将其发送到 ION-SFU（选择性转发单元），从而使 WebRTC 会话更有效地扩展。

WebRTC 是 Web Real-Time Communication 的缩写，是一种利用点对点连接在网络上实现实时音频、视频和数据传输的通信协议。

WebRTC 还提供了大多数浏览器默认提供的 Javascript API，可帮助开发人员在其应用程序中实现该协议。但是也有一些其他语言的 WebRTC 协议的实现。

在本教程中，您将构建一个视频广播应用程序，该应用程序在 Golang 中读取摄像头并将其发送到 ION-SFU（选择性转发单元），从而使 WebRTC 会话更有效地扩展。

该应用程序还将配备一个小型前端，让您可以通过从 ION-SFU 服务器读取您发布的视频来观看它。

目录
先决条件
技术栈
设置 ION-SFU
创建项目
建立 WebRTC 连接
客户端
Ion-SDK-Go
结论
先决条件
在开始本指南之前，您需要具备以下条件：

有效的 Golang 安装。
连接到计算机的摄像头，可以使用 Video for Linux 作为视频流的来源进行读取。
（可选）如果你想连接不在你网络上的设备，你需要在你的应用程序中添加一个 TURN 服务器。如果您想了解更多关于 TURN 的信息以及如何设置您自己的 TURN。
技术栈
现在您已经大致了解了要构建的内容，让我们仔细看看正在使用的工具以及它们如何相互协作。

让我们分解不同的组件：

Pion -WebRTC 协议的纯 Golang 实现。用于与 ION-SFU 建立对等连接并发送视频流。
ION SFU -ION SFU（选择性转发单元）是一种视频路由服务，可让 Webrtc 会话更有效地扩展。
Pion mediadevices – Mediadevices API的 Golang 实现，用于将相机读取为可以使用对等连接发送的媒体流。
这样做的一个主要好处是您无需打开浏览器选项卡即可读取相机。使用选择性转发单元也将有助于提高性能并为大量用户扩展应用程序。

设置 ION-SFU
在本节中，您将克隆和配置 ION-SFU 服务器，以便您可以将其用于您的应用程序。

首先，您将克隆存储库，以便拥有开始设置选择性转发单元所需的所有资源：

git clone --branch v1.10.6 https://github.com/pion/ion-sfu.git
此命令将从 Github 克隆 ION-SFU 存储库，并在您的目录中创建一个名为ion-sfu的文件夹。现在使用以下命令进入目录：

cd ion-sfu
接下来，您可以通过更改config.toml文件来编辑 sfu 的配置。标准配置适合测试和本地使用，但如果您尝试从另一个网络中的设备访问服务器，我建议添加 STUN 和 TURN 服务器。

完成配置后，您可以使用以下命令启动服务器：

go build ./cmd/signal/json-rpc/main.go && ./main -c config.toml

  • 1

或者，如果您更喜欢使用 Golang 启动服务器，也可以使用 Docker 启动服务器。

docker run -p 7000:7000 -p 5000-5020:5000-5020/udp pionwebrtc/ion-sfu:v1.10.6-jsonrpc

  • 1

您现在已经成功设置了 ION-SFU 服务器，应该会在控制台中看到以下输出。

config config.toml load ok!
[2020-10-12 19:04:19.017] [INFO] [376][main.go][main] => — Starting SFU Node —
[2020-10-12 19:04:19.018] [INFO] [410][main.go][main] => Listening at http://[:7000]
创建项目
现在 ion-sfu 服务器的设置和配置已经完成，是时候创建项目了

首先，您需要创建一个目录并进入该目录。

mkdir mediadevice-broadcast && cd mediadevice-broadcast
之后，您可以使用以下命令继续创建项目所需的所有文件：

mkdir public
touch main.go public/index.html public/index.js public/style.css
跟随本文还需要安装两个包。

sudo apt-get install -y v4l-utils
sudo apt-get install -y libvpx-dev
如果您不在 Linux 上，则可能需要下载不同的软件包。查看媒体设备文档以获取更多信息。

建立 WebRTC 连接
在使用 WebRTC 交换任何数据之前，必须首先在两个 WebRTC 代理之间建立对等连接。由于点对点连接通常不能直接建立，因此需要一些信令方法。

发送给 ion-sfu 的信号将通过 Websockets 协议处理。为此，我们将使用连接到 Websockets 服务器的gorilla/websocket库实现一个简单的 Websockets 样板，并允许我们接收传入消息并发送我们自己的消息。

p

ackage main

import (
	"bytes"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"net/url"

	"github.com/google/uuid"
	"github.com/gorilla/websocket"
)

var addr string

func main() {
	flag.StringVar(&addr, "a", "localhost:7000", "address to use")
	flag.Parse()

	u := url.URL{Scheme: "ws", Host: addr, Path: "/ws"}
	log.Printf("connecting to %s", u.String())

	c, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
	if err != nil {
		log.Fatal("dial:", err)
	}
	defer c.Close()

	// Read incoming Websocket messages
	done := make(chan struct{})

	go readMessage(c, done)

	<-done
}

func readMessage(connection *websocket.Conn, done chan struct{}) {
	defer close(done)
	for {
		_, message, err := connection.ReadMessage()
		if err != nil || err == io.EOF {
			log.Fatal("Error reading: ", err)
			break
		}

		fmt.Printf("recv: %s", message)
	}
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50

现在让我们浏览一下代码以便更好地理解：

该标志用于在启动脚本时动态提供 Websockets 服务器的 URL，标准值为localhost:7000
该 URL 用于使用Dial方法创建 Websockets 客户端。然后我们检查连接是否导致错误，如果是这样则打印日志。
readMessage函数然后通过在 Websocket 连接上调用ReadMessage()来读取传入的消息，并作为 Go 例程运行，因此它不会阻塞主线程并且可以在后台运行。
main()函数的最后一行确保脚本在done变量未关闭时运行。
下一步是创建到 ion-sfu 的对等连接并处理传入的 WebRTC 信号事件。

var peerConnection *webrtc.PeerConnection

func main() {
...

    config := webrtc.Configuration{
		ICEServers: []webrtc.ICEServer{
			{
				URLs: []string{"stun:stun.l.google.com:19302"},
			},
			/*{
				URLs:       []string{"turn:TURN_IP:3478?transport=tcp"},
				Username:   "username",
				Credential: "password",
			},*/
		},
		SDPSemantics: webrtc.SDPSemanticsUnifiedPlanWithFallback,
	}

	// Create a new RTCPeerConnection
	mediaEngine := webrtc.MediaEngine{}

	vpxParams, err := vpx.NewVP8Params()
	if err != nil {
		panic(err)
	}
	vpxParams.BitRate = 500_000 // 500kbps

	codecSelector := mediadevices.NewCodecSelector(
		mediadevices.WithVideoEncoders(&vpxParams),
	)

	codecSelector.Populate(&mediaEngine)
	api := webrtc.NewAPI(webrtc.WithMediaEngine(mediaEngine))
	peerConnection, err = api.NewPeerConnection(config)
	if err != nil {
		panic(err)
	}

}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40

在这里，我们首先创建一个 WebRTC 配置，我们在其中定义将在信令过程中使用的 STUN 和 TURN 服务器。之后，我们创建一个MediaEngine，让我们定义对等连接支持的编解码器。

完成所有这些配置后，我们可以通过在我们刚刚创建的 WebRTC API 上调用NewPeerConnection函数来创建新的对等连接。

在通过 Websockets 将报价发送到 ion-sfu 服务器之前，我们首先需要添加视频和音频流。这是媒体设备库发挥作用以从摄像机读取视频的地方。

 fmt.Println(mediadevices.EnumerateDevices())

	s, err := mediadevices.GetUserMedia(mediadevices.MediaStreamConstraints{
		Video: func(c *mediadevices.MediaTrackConstraints) {
			c.FrameFormat = prop.FrameFormat(frame.FormatYUY2)
			c.Width = prop.Int(640)
			c.Height = prop.Int(480)
		},
		Codec: codecSelector,
	})

	if err != nil {
		panic(err)
	}

	for _, track := range s.GetTracks() {
		track.OnEnded(func(err error) {
			fmt.Printf("Track (ID: %s) ended with error: %v\n",
				track.ID(), err)
		})
		_, err = peerConnection.AddTransceiverFromTrack(track,
			webrtc.RtpTransceiverInit{
				Direction: webrtc.RTPTransceiverDirectionSendonly,
			},
		)
		if err != nil {
			panic(err)
		}
	}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29

使用对等连接创建媒体设备库的实例后，您可以使用GetUserMedia函数并传递参数来获取用户媒体。

您可能需要进行的一项配置更改是更改 FrameFormat以支持您连接的相机。您可以使用以下命令检查相机的帧格式：

v4l2-ctl --all
所有支持的格式也可以在媒体设备 Github 存储库中找到。

现在可以创建报价并将其保存到对等连接的本地描述中。

// Creating WebRTC offer
offer, err := peerConnection.CreateOffer(nil)

// Set the remote SessionDescription
err = peerConnection.SetLocalDescription(offer)
if err != nil {
panic(err)
}

下一步是使用 Websockets 将报价发送到 sfu。Websockets 消息是 JSON，需要特定的结构才能被 sfu 识别。

因此，我们需要创建一个结构来保存我们的报价和指定我们想要加入的房间的所需 sid，然后我们可以将其转换为 JSON。

type SendOffer struct {
	SID   string                     `json:sid`
	Offer *webrtc.SessionDescription `json:offer`
}

  • 1
  • 2
  • 3
  • 4

现在我们使用json.Marshal()函数将报价对象转换为 JSON ，然后使用 JSON 报价对象作为请求中的参数。

将请求转换为字节数组后，消息最终可以使用WriteMessage()函数通过 Websockets 发送。

offerJSON, err := json.Marshal(&SendOffer{
Offer: peerConnection.LocalDescription(),
SID: “test room”,
})

params := (*json.RawMessage)(&offerJSON)

connectionUUID := uuid.New()
connectionID = uint64(connectionUUID.ID())

offerMessage := &jsonrpc2.Request{
Method: “join”,
Params: params,
ID: jsonrpc2.ID{
IsString: false,
Str: “”,
Num: connectionID,
},
}

reqBodyBytes := new(bytes.Buffer)
json.NewEncoder(reqBodyBytes).Encode(offerMessage)

messageBytes := reqBodyBytes.Bytes()
c.WriteMessage(websocket.TextMessage, messageBytes)

现在报价已发送，我们需要正确响应 WebRTC 事件和来自 Websockets 服务器的响应。

每当找到新的 ICE 候选对象时，都会调用OnICECandidate事件。然后使用该方法通过向 sfu 发送 trickle 请求来协商与远程对等方的连接。

// Handling OnICECandidate event
peerConnection.OnICECandidate(func(candidate *webrtc.ICECandidate) {
if candidate != nil {
candidateJSON, err := json.Marshal(&Candidate{
Candidate: candidate,
Target: 0,
})

	params := (*json.RawMessage)(&candidateJSON)

	if err != nil {
		log.Fatal(err)
	}

	message := &jsonrpc2.Request{
		Method: "trickle",
		Params: params,
	}

	reqBodyBytes := new(bytes.Buffer)
	json.NewEncoder(reqBodyBytes).Encode(message)

	messageBytes := reqBodyBytes.Bytes()
	c.WriteMessage(websocket.TextMessage, messageBytes)
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17

})

peerConnection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
fmt.Printf(“Connection State has changed to %s \n”, connectionState.String())
})

之前创建的readMessage函数用于接收和响应 sfu 发送的传入 Websockets 消息。

为此，我们首先需要创建包含接收到的消息的结构，以便我们可以使用数据。然后我们将确定消息是针对哪个事件并相应地处理它们。

// SendAnswer object to send to the sfu over Websockets
type SendAnswer struct {
	SID    string                     `json:sid`
	Answer *webrtc.SessionDescription `json:answer`
}

type ResponseCandidate struct {
	Target    int                   `json:"target"`
	Candidate *webrtc.ICECandidateInit `json:candidate`
}

// TrickleResponse received from the sfu server
type TrickleResponse struct {
	Params ResponseCandidate				`json:params`
	Method string                   `json:method`
}

// Response received from the sfu over Websockets
type Response struct {
	Params *webrtc.SessionDescription `json:params`
	Result *webrtc.SessionDescription `json:result`
	Method string                     `json:method`
	Id     uint64                     `json:id`
}

func readMessage(connection *websocket.Conn, done chan struct{}) {
	defer close(done)
	for {
		_, message, err := connection.ReadMessage()
		if err != nil || err == io.EOF {
			log.Fatal("Error reading: ", err)
			break
		}

		fmt.Printf("recv: %s", message)

		var response Response
		json.Unmarshal(message, &response)

		if response.Id == connectionID {
			result := *response.Result
			remoteDescription = response.Result
			if err := peerConnection.SetRemoteDescription(result); err != nil {
				log.Fatal(err)
			}
		} else if response.Id != 0 && response.Method == "offer" {
			peerConnection.SetRemoteDescription(*response.Params)
			answer, err := peerConnection.CreateAnswer(nil)

			if err != nil {
				log.Fatal(err)
			}

			peerConnection.SetLocalDescription(answer)

			connectionUUID := uuid.New()
			connectionID = uint64(connectionUUID.ID())

			offerJSON, err := json.Marshal(&SendAnswer{
				Answer: peerConnection.LocalDescription(),
				SID:    "test room",
			})

			params := (*json.RawMessage)(&offerJSON)

			answerMessage := &jsonrpc2.Request{
				Method: "answer",
				Params: params,
				ID: jsonrpc2.ID{
					IsString: false,
					Str:      "",
					Num:      connectionID,
				},
			}

			reqBodyBytes := new(bytes.Buffer)
			json.NewEncoder(reqBodyBytes).Encode(answerMessage)

			messageBytes := reqBodyBytes.Bytes()
			connection.WriteMessage(websocket.TextMessage, messageBytes)
		} else if response.Method == "trickle" {
			var trickleResponse TrickleResponse

			if err := json.Unmarshal(message, &trickleResponse); err != nil {
				log.Fatal(err)
			}

			err := peerConnection.AddICECandidate(*trickleResponse.Params.Candidate)

			if err != nil {
				log.Fatal(err)
			}
		}
	}
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95

如您所见，我们正在处理两个不同的事件：

Offer – sfu 发送一个 offer，我们通过将发送的 offer 保存到我们对等连接的远程描述中并发回一个带有本地描述的答案来做出反应，这样我们就可以连接到远程对等点。
涓流 – sfu 发送一个新的 ICE 候选者，我们将其添加到对等连接
所有这些配置将产生以下文件：

package main

import (
	"bytes"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"net/url"

	"github.com/google/uuid"
	"github.com/gorilla/websocket"
	"github.com/pion/mediadevices"
	"github.com/pion/mediadevices/pkg/codec/vpx"
	"github.com/pion/mediadevices/pkg/frame"
	"github.com/pion/mediadevices/pkg/prop"
	"github.com/pion/webrtc/v3"
	"github.com/sourcegraph/jsonrpc2"

	// Note: If you don't have a camera or microphone or your adapters are not supported,
	//       you can always swap your adapters with our dummy adapters below.
	// _ "github.com/pion/mediadevices/pkg/driver/videotest"
	// _ "github.com/pion/mediadevices/pkg/driver/audiotest"
	_ "github.com/pion/mediadevices/pkg/driver/camera"     // This is required to register camera adapter
	_ "github.com/pion/mediadevices/pkg/driver/microphone" // This is required to register microphone adapter
)

type Candidate struct {
	Target    int                   `json:"target"`
	Candidate *webrtc.ICECandidate `json:candidate`
}

type ResponseCandidate struct {
	Target    int                   `json:"target"`
	Candidate *webrtc.ICECandidateInit `json:candidate`
}

// SendOffer object to send to the sfu over Websockets
type SendOffer struct {
	SID   string                     `json:sid`
	Offer *webrtc.SessionDescription `json:offer`
}

// SendAnswer object to send to the sfu over Websockets
type SendAnswer struct {
	SID    string                     `json:sid`
	Answer *webrtc.SessionDescription `json:answer`
}

// TrickleResponse received from the sfu server
type TrickleResponse struct {
	Params ResponseCandidate				`json:params`
	Method string                   `json:method`
}

// Response received from the sfu over Websockets
type Response struct {
	Params *webrtc.SessionDescription `json:params`
	Result *webrtc.SessionDescription `json:result`
	Method string                     `json:method`
	Id     uint64                     `json:id`
}

var peerConnection *webrtc.PeerConnection
var connectionID uint64
var remoteDescription *webrtc.SessionDescription

var addr string

func main() {
	flag.StringVar(&addr, "a", "localhost:7000", "address to use")
	flag.Parse()

	u := url.URL{Scheme: "ws", Host: addr, Path: "/ws"}
	log.Printf("connecting to %s", u.String())

	c, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
	if err != nil {
		log.Fatal("dial:", err)
	}
	defer c.Close()

	config := webrtc.Configuration{
		ICEServers: []webrtc.ICEServer{
			{
				URLs: []string{"stun:stun.l.google.com:19302"},
			},
			/*{
				URLs:       []string{"turn:TURN_IP:3478"},
				Username:   "username",
				Credential: "password",
			},*/
		},
		SDPSemantics: webrtc.SDPSemanticsUnifiedPlanWithFallback,
	}

	// Create a new RTCPeerConnection
	mediaEngine := webrtc.MediaEngine{}

	vpxParams, err := vpx.NewVP8Params()
	if err != nil {
		panic(err)
	}
	vpxParams.BitRate = 500_000 // 500kbps

	codecSelector := mediadevices.NewCodecSelector(
		mediadevices.WithVideoEncoders(&vpxParams),
	)

	codecSelector.Populate(&mediaEngine)
	api := webrtc.NewAPI(webrtc.WithMediaEngine(&mediaEngine))
	peerConnection, err = api.NewPeerConnection(config)
	if err != nil {
		panic(err)
	}

	// Read incoming Websocket messages
	done := make(chan struct{})

	go readMessage(c, done)

	fmt.Println(mediadevices.EnumerateDevices())

	s, err := mediadevices.GetUserMedia(mediadevices.MediaStreamConstraints{
		Video: func(c *mediadevices.MediaTrackConstraints) {
			c.FrameFormat = prop.FrameFormat(frame.FormatYUY2)
			c.Width = prop.Int(640)
			c.Height = prop.Int(480)
		},
		Codec: codecSelector,
	})

	if err != nil {
		panic(err)
	}

	for _, track := range s.GetTracks() {
		track.OnEnded(func(err error) {
			fmt.Printf("Track (ID: %s) ended with error: %v\n",
				track.ID(), err)
		})
		_, err = peerConnection.AddTransceiverFromTrack(track,
			webrtc.RtpTransceiverInit{
				Direction: webrtc.RTPTransceiverDirectionSendonly,
			},
		)
		if err != nil {
			panic(err)
		}
	}

	// Creating WebRTC offer
	offer, err := peerConnection.CreateOffer(nil)

	// Set the remote SessionDescription
	err = peerConnection.SetLocalDescription(offer)
	if err != nil {
		panic(err)
	}

	// Handling OnICECandidate event
	peerConnection.OnICECandidate(func(candidate *webrtc.ICECandidate) {
		if candidate != nil {
			candidateJSON, err := json.Marshal(&Candidate{
				Candidate: candidate,
				Target: 0,
			})

			params := (*json.RawMessage)(&candidateJSON)

			if err != nil {
				log.Fatal(err)
			}

			message := &jsonrpc2.Request{
				Method: "trickle",
				Params: params,
			}

			reqBodyBytes := new(bytes.Buffer)
			json.NewEncoder(reqBodyBytes).Encode(message)

			messageBytes := reqBodyBytes.Bytes()
			c.WriteMessage(websocket.TextMessage, messageBytes)
		}
	})

	peerConnection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
		fmt.Printf("Connection State has changed to %s \n", connectionState.String())
	})

	offerJSON, err := json.Marshal(&SendOffer{
		Offer: peerConnection.LocalDescription(),
		SID:   "test room",
	})

	params := (*json.RawMessage)(&offerJSON)

	connectionUUID := uuid.New()
	connectionID = uint64(connectionUUID.ID())

	offerMessage := &jsonrpc2.Request{
		Method: "join",
		Params: params,
		ID: jsonrpc2.ID{
			IsString: false,
			Str:      "",
			Num:      connectionID,
		},
	}

	reqBodyBytes := new(bytes.Buffer)
	json.NewEncoder(reqBodyBytes).Encode(offerMessage)

	messageBytes := reqBodyBytes.Bytes()
	c.WriteMessage(websocket.TextMessage, messageBytes)

	<-done
}

func readMessage(connection *websocket.Conn, done chan struct{}) {
	defer close(done)
	for {
		_, message, err := connection.ReadMessage()
		if err != nil || err == io.EOF {
			log.Fatal("Error reading: ", err)
			break
		}

		fmt.Printf("recv: %s", message)

		var response Response
		json.Unmarshal(message, &response)

		if response.Id == connectionID {
			result := *response.Result
			remoteDescription = response.Result
			if err := peerConnection.SetRemoteDescription(result); err != nil {
				log.Fatal(err)
			}
		} else if response.Id != 0 && response.Method == "offer" {
			peerConnection.SetRemoteDescription(*response.Params)
			answer, err := peerConnection.CreateAnswer(nil)

			if err != nil {
				log.Fatal(err)
			}

			peerConnection.SetLocalDescription(answer)

			connectionUUID := uuid.New()
			connectionID = uint64(connectionUUID.ID())

			offerJSON, err := json.Marshal(&SendAnswer{
				Answer: peerConnection.LocalDescription(),
				SID:    "test room",
			})

			params := (*json.RawMessage)(&offerJSON)

			answerMessage := &jsonrpc2.Request{
				Method: "answer",
				Params: params,
				ID: jsonrpc2.ID{
					IsString: false,
					Str:      "",
					Num:      connectionID,
				},
			}

			reqBodyBytes := new(bytes.Buffer)
			json.NewEncoder(reqBodyBytes).Encode(answerMessage)

			messageBytes := reqBodyBytes.Bytes()
			connection.WriteMessage(websocket.TextMessage, messageBytes)
		} else if response.Method == "trickle" {
			var trickleResponse TrickleResponse

			if err := json.Unmarshal(message, &trickleResponse); err != nil {
				log.Fatal(err)
			}

			err := peerConnection.AddICECandidate(*trickleResponse.Params.Candidate)

			if err != nil {
				log.Fatal(err)
			}
		}
	}
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117
  • 118
  • 119
  • 120
  • 121
  • 122
  • 123
  • 124
  • 125
  • 126
  • 127
  • 128
  • 129
  • 130
  • 131
  • 132
  • 133
  • 134
  • 135
  • 136
  • 137
  • 138
  • 139
  • 140
  • 141
  • 142
  • 143
  • 144
  • 145
  • 146
  • 147
  • 148
  • 149
  • 150
  • 151
  • 152
  • 153
  • 154
  • 155
  • 156
  • 157
  • 158
  • 159
  • 160
  • 161
  • 162
  • 163
  • 164
  • 165
  • 166
  • 167
  • 168
  • 169
  • 170
  • 171
  • 172
  • 173
  • 174
  • 175
  • 176
  • 177
  • 178
  • 179
  • 180
  • 181
  • 182
  • 183
  • 184
  • 185
  • 186
  • 187
  • 188
  • 189
  • 190
  • 191
  • 192
  • 193
  • 194
  • 195
  • 196
  • 197
  • 198
  • 199
  • 200
  • 201
  • 202
  • 203
  • 204
  • 205
  • 206
  • 207
  • 208
  • 209
  • 210
  • 211
  • 212
  • 213
  • 214
  • 215
  • 216
  • 217
  • 218
  • 219
  • 220
  • 221
  • 222
  • 223
  • 224
  • 225
  • 226
  • 227
  • 228
  • 229
  • 230
  • 231
  • 232
  • 233
  • 234
  • 235
  • 236
  • 237
  • 238
  • 239
  • 240
  • 241
  • 242
  • 243
  • 244
  • 245
  • 246
  • 247
  • 248
  • 249
  • 250
  • 251
  • 252
  • 253
  • 254
  • 255
  • 256
  • 257
  • 258
  • 259
  • 260
  • 261
  • 262
  • 263
  • 264
  • 265
  • 266
  • 267
  • 268
  • 269
  • 270
  • 271
  • 272
  • 273
  • 274
  • 275
  • 276
  • 277
  • 278
  • 279
  • 280
  • 281
  • 282
  • 283
  • 284
  • 285
  • 286
  • 287
  • 288
  • 289
  • 290
  • 291

注意：您可能需要启用 go modules 以便在启动脚本时自动下载依赖项。

现在可以使用以下命令启动完成的脚本：

You might need to add sudo to access your camera

go run main.go
您应该看到以下输出：

recv: {"method":"trickle","params":{"candidate":"candidate:3681230645 1 udp 2130706431 10.0.0.35 49473 typ host","sdpMid":"","sdpMLineIndex":0,"usernameFragment":null},"jsonrpc":"2.0"}
recv: {"method":"trickle","params":{"candidate":"candidate:233762139 1 udp 2130706431 172.17.0.1 57218 typ host","sdpMid":"","sdpMLineIndex":0,"usernameFragment":null},"jsonrpc":"2.0"}
Connection State has changed to checking 
recv: {"method":"trickle","params":{"candidate":"candidate:2890797847 1 udp 2130706431 172.22.0.1 41179 typ host","sdpMid":"","sdpMLineIndex":0,"usernameFragment":null},"jsonrpc":"2.0"}
recv: {"method":"trickle","params":{"candidate":"candidate:3528925834 1 udp 2130706431 172.18.0.1 58906 typ host","sdpMid":"","sdpMLineIndex":0,"usernameFragment":null},"jsonrpc":"2.0"}
recv: {"method":"trickle","params":{"candidate":"candidate:3197649470 1 udp 1694498815 212.197.155.248 36942 typ srflx raddr 0.0.0.0 rport 36942","sdpMid":"","sdpMLineIndex":0,"usernameFragment":null},"jsonrpc":"2.0"}
recv: {"method":"trickle","params":{"candidate":"candidate:2563076625 1 udp 16777215 104.248.140.156 11643 typ relay raddr 0.0.0.0 rport 42598","sdpMid":"","sdpMLineIndex":0,"usernameFragment":null},"jsonrpc":"2.0"}
Connection State has changed to connected 

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

客户端
现在我们已经成功地将视频从摄像机发送到 sfu，是时候创建一个前端来接收它了。

HTML 文件非常基本，只包含一个视频对象和一个订阅流的按钮。它还会将当前的 WebRTC 日志打印到一个 div 中。

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="utf-8"/>
    <meta
            name="viewport"
            content="width=device-width, initial-scale=1, shrink-to-fit=no"
    />

    <style>
        #remotes video {
            width: 320px;
        }
    </style>

    <title>WebRTC test frontend</title>
</head>
<body>
<div>
    <div id="remotes">
          <span
                  style="position: absolute; margin-left: 5px; margin-top: 5px"
                  class="badge badge-primary"
          >Remotes</span
          >
    </div>
</div>

<script src="https://unpkg.com/ion-sdk-js@1.5.5/dist/ion-sdk.min.js"></script>
<script src="https://unpkg.com/ion-sdk-js@1.5.5/dist/json-rpc.min.js"></script>
<script src="index.js"></script>
</body>
</html>

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33

然后 javascript 文件将连接到 sfu，类似于上面的 Golang 脚本。唯一的区别是，它不是读取相机并将视频发送到 sfu，而是接收视频。

我不会详细介绍，因为上面已经涵盖了所有功能。

const remotesDiv = document.getElementById("remotes");

const config = {
    codec: 'vp8',
    iceServers: [
        {
            "urls": "stun:stun.l.google.com:19302",
        },
        /*{
            "urls": "turn:TURN_IP:3468",
            "username": "username",
            "credential": "password"
        },*/
    ]
};

const signalLocal = new Signal.IonSFUJSONRPCSignal(
    "ws://127.0.0.1:7000/ws"
);

const clientLocal = new IonSDK.Client(signalLocal, config);
signalLocal.onopen = () => clientLocal.join("test room");

clientLocal.ontrack = (track, stream) => {
    console.log("got track", track.id, "for stream", stream.id);
    if (track.kind === "video") {
        track.onunmute = () => {
            const remoteVideo = document.createElement("video");
            remoteVideo.srcObject = stream;
            remoteVideo.autoplay = true;
            remoteVideo.muted = true;
            remotesDiv.appendChild(remoteVideo);

            track.onremovetrack = () => remotesDiv.removeChild(remoteVideo);
        };
    }
};

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37

在这里您唯一需要记住的是，如果不提供或请求某种流，就无法发送对等连接。这就是为什么在发送报价之前添加两个接收器（一个用于音频，一个用于视频）。

您现在可以通过在浏览器中打开 HTML 文件来启动前端。或者，您可以通过在项目的根目录中创建一个新文件来使用 Express 服务器打开 HTML 文件。

touch server.js
在添加代码之前需要安装 express 依赖项。

npm init -y
npm install express --save
然后您可以使用以下代码将前端作为静态站点运行。

const express = require("express");
const app = express();

const port = 3000;

const http = require("http");
const server = http.createServer(app);

app.use(express.static(__dirname + "/public"));

server.listen(port, () => console.log(`Server is running on port ${port}`));

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

使用以下命令启动应用程序。

node server.js

  • 1

您现在应该能够访问本地计算机的 localhost:3000 上的前端。

Ion-SDK-Go
如上所述，同样的视频流功能也可以使用ion 团队创建的库来实现，它抽象了 WebRTC 信号，因此使实现更短、更简洁。不过，知道如何自己实现信号非常重要，并且可以让您为更复杂的项目进行更多定制。

package main

import (
	"flag"
	"fmt"

	ilog "github.com/pion/ion-log"
	sdk "github.com/pion/ion-sdk-go"
	"github.com/pion/mediadevices"
	"github.com/pion/mediadevices/pkg/codec/vpx"
	"github.com/pion/mediadevices/pkg/frame"
	"github.com/pion/mediadevices/pkg/prop"
	"github.com/pion/webrtc/v3"

	// Note: If you don't have a camera or microphone or your adapters are not supported,
	//       you can always swap your adapters with our dummy adapters below.
	// _ "github.com/pion/mediadevices/pkg/driver/videotest"
	// _ "github.com/pion/mediadevices/pkg/driver/audiotest"
	_ "github.com/pion/mediadevices/pkg/driver/camera"     // This is required to register camera adapter
	_ "github.com/pion/mediadevices/pkg/driver/microphone" // This is required to register microphone adapter
)

var (
	log = ilog.NewLoggerWithFields(ilog.DebugLevel, "", nil)
)

func main() {

	// parse flag
	var session, addr string
	flag.StringVar(&addr, "addr", "localhost:50051", "Ion-sfu grpc addr")
	flag.StringVar(&session, "session", "test room", "join session name")
	flag.Parse()

	// add stun servers
	webrtcCfg := webrtc.Configuration{
		ICEServers: []webrtc.ICEServer{
			webrtc.ICEServer{
				URLs: []string{"stun:stun.stunprotocol.org:3478", "stun:stun.l.google.com:19302"},
			},
		},
	}

	config := sdk.Config{
		Log: log.Config{
			Level: "debug",
		},
		WebRTC: sdk.WebRTCTransportConfig{
			Configuration: webrtcCfg,
		},
	}
	// new sdk engine
	e := sdk.NewEngine(config)

	// get a client from engine
	c, err := sdk.NewClient(e, addr, "client id")

	c.GetPubTransport().GetPeerConnection().OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
		log.Infof("Connection state changed: %s", state)
	})

	if err != nil {
		log.Errorf("client err=%v", err)
		panic(err)
	}

	e.AddClient(c)

	// client join a session
	err = c.Join(session, nil)

	if err != nil {
		log.Errorf("join err=%v", err)
		panic(err)
	}

	vpxParams, err := vpx.NewVP8Params()
	if err != nil {
		panic(err)
	}
	vpxParams.BitRate = 500_000 // 500kbps

	codecSelector := mediadevices.NewCodecSelector(
		mediadevices.WithVideoEncoders(&vpxParams),
	)

	fmt.Println(mediadevices.EnumerateDevices())

	s, err := mediadevices.GetUserMedia(mediadevices.MediaStreamConstraints{
		Video: func(c *mediadevices.MediaTrackConstraints) {
			c.FrameFormat = prop.FrameFormat(frame.FormatYUY2)
			c.Width = prop.Int(640)
			c.Height = prop.Int(480)
		},
		Codec: codecSelector,
	})

	if err != nil {
		panic(err)
	}

	for _, track := range s.GetTracks() {
		track.OnEnded(func(err error) {
			fmt.Printf("Track (ID: %s) ended with error: %v\n",
				track.ID(), err)
		})
		_, err = c.Publish(track)
		if err != nil {
			panic(err)
		} else {
			break // only publish first track, thanks
		}
	}

	select {}
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116

如您所见，该库处理信号，因此抽象了我们上面编写的大量代码。该库与我们上面实现的代码之间的一个区别是该库使用 GRPC 进行信号传输，而我们使用 JSONRPC。因此，您必须以 AllRPC 模式而不是 JSONRPC 模式启动 ion-sfu。这可以通过使用 Golang 启动 AllRPC 版本或在使用 Docker 时使用AllRPC 图像标签（例如 latest-allrpc）来完成。

您现在可以使用 go run 命令启动应用程序：

go run ./main.go
您现在应该还可以在本地机器的 localhost:3000 上看到您的摄像头的视频。

结论
在本文中，您了解了什么是 sfu 以及如何利用 ion-sfu 构建视频广播应用程序。您还学习了如何使用 Golang 媒体设备库在不打开浏览器窗口的情况下读取您的摄像头。

作者：Gabriel Tanner
原文链接：https://gabrieltanner.org/blog/broadcasting-ion-sfu/

https://www.nxrte.com/jishu/webrtc/14792.html

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小蓝xlanll/article/detail/157527
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号