资料库的webrtc文件传输

本文主要是介绍资料库的webrtc文件传输，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

一、一个看似简单的事情往往不简单

一个简单的事情往往会倾注你的心血，也许你看到很简单往往其实没那么简单；其实想想今年业余时间的大把代码，真正能成品的好像并不多。

马上年底了，写下这篇文章。每一行程序就像写小说一样，不管好不好；代码都倾注了我的心血。

真正把一个东西做好；做到好用，用的人多，具有很好的利他性，并不是一个容易的事情，需要不断的量变。

webrtc演示网址:

文件下载测试

二、为什么放弃百度网盘

把数据存储到百度网盘上大该有半年时间，后来发现了一些致命的问题

1、发现分享外链到了10几万条之后就不能分享了

2、分享的百度外链经常会被屏蔽掉

3、分享速度不能过快，1小时只能分享100多条外链，如果分享的过快整个网盘的分享功能都会

4、百度账号需要实名认证，很容易被人认出来，作为一个技术人员我认为应该藏在幕后，尽量不要让别人认出来，这回避免很多不必要的问题

三、为什么不用云存储

4个T的存储，每月就要400元，一年就是5600元，而且不算流量费用一年就是5600，这个费用是不高的，最关键的是使用了云存储很

1、可能被黑客共计，如果别人使用curl去恶意刷流量现有的安全策略很可能被人把费用都刷去

2、每年存储费用是递增的

3、需要实名认证，一旦绑定域名，绑定着手机号，很可能被人找到

四、我的WebRtc架构设计

拜托传统的中继服务器，使用p2p打洞，调研了一些方案，本来打算使用 coturn + janus 的架构设计，但是发现janus c 写的，没有文件传输的插件，如果自己再用c去做一个插件真的是很累，于是就打算用golang 实现一个文件传输的网关。

大体步骤

1、启动网关

2、存储节点连接到网关

3、用户浏览器请求网关，请求和存储节点交换信令

4、信令交换完成，使用dataChannel进行通信

五、网关设计

网关主要是一个websocket server 采用golang编写

网关设计主要分成了几大模块

1、room

room是一个公共领域，主要是做client和node的鉴权操作，如果鉴权成功，那么会进入Manager进行调度

2、抽象client和node

client和node 有一个共性是他们都具有连接属性，所以应该设计一个公共接口，把他们的共性抽象出来

type Lifecycle interface {Stop()Start()
}type Describable interface {Name() stringType() ConnType
}type Component interface {LifecycleDescribableAddCallback(ctx EventCallback)GetConnNumber() uint64SetContext(interface{})GetContext() interface{}Product(ctx Context)
}

通过添加一个AddCallback回调函数，确保不同模块的业务处理完全玻璃开，node的逻辑就在node里处理，client的逻辑只在client里处理，不能把不同模块的代码交叉处理，至于上下问的传输，统一抽象一个Context，里面抽象存储着我们需要的上下文信息，供给不同的回调函数以及协成之间传输使用。

type Context interface {GetData() []byteError() errorMessageType() intGetContext() interface{}SetContext(interface{})Name() string
}type EventCallback interface {OnReceive(Context, *sync.WaitGroup)OnError(Context, *sync.WaitGroup)OnClose(Context, *sync.WaitGroup)OnWriteComplete(Context, *sync.WaitGroup)
}type NodeCallback interface {OnReceive(Context, *sync.WaitGroup)OnError(Context, *sync.WaitGroup)OnClose(Context, *sync.WaitGroup)OnWriteComplete(Context, *sync.WaitGroup)
}

4、manager

负责整体调度，比如client进入manager后，查找当前可以用的node存储节点，找到后进行信令交换

节点调度，遍历查找拥有client最少的node节点，然后进行通信

// 选择最优线路
func (m *Manager) selectNode() *node.NodeClient {if len(m.nodeTree) == 0 {return nil}// 找一个挂载链接最少的节点var usableNode *node.NodeClientvar weight uint64for _, conn := range m.nodeTree {if uint64(conn.GetConnNumber()) <= weight {usableNode = conn.(*node.NodeClient)}}return usableNode
}

六、node节点收到信令后进行应答

1、创建RTCConnection

blockSize := 16384//前端页面会对sdp进行base64的encodeb, err := base64.StdEncoding.DecodeString(sdp)if err != nil {log.Error("error:%s", err)return nil}str, err := url.QueryUnescape(string(b))if err != nil {log.Error("error:%s", err)return nil}sdpDes := webrtc.SessionDescription{}fmt.Println(str)err = json.Unmarshal([]byte(str), &sdpDes)if err != nil {log.Error("json.Unmarshal err:%s", err)return nil}//创建pc， 并且指定stun服务器pc, err := webrtc.NewPeerConnection(webrtc.Configuration{ICEServers: []webrtc.ICEServer{{URLs: []string{"stun:"},},},})stat, err := os.Stat("/home/zhanglei/Downloads/《实践论》(原文)毛泽东.pdf")if err != nil {log.Error("os.Stat %s error:%s", path, err)return nil}if offset > stat.Size() {log.Error("offset(%d) > stat.Size(%d)", offset, stat.Size())return nil}chunkSize := int(math.Ceil(float64(stat.Size() / int64(blockSize))))currentChunkSize := int(math.Ceil(float64(offset / int64(blockSize))))if err != nil {log.Error("%s", err)return nil}pc.OnConnectionStateChange(func(s webrtc.PeerConnectionState) {fmt.Println("OnConnectionStateChange")fmt.Printf("Peer Connection State has changed: %s (answerer)\n", s.String())if s == webrtc.PeerConnectionStateFailed {fmt.Println("webrtc.PeerConnectionStateFailed")}})// Register data channel creation handlingpc.OnDataChannel(func(d *webrtc.DataChannel) {fmt.Printf("New DataChannel %s %d\n", d.Label(), d.ID())// Register channel opening handlingd.OnOpen(func() {fmt.Printf("Data channel '%s'-'%d' open. Random messages will now be sent to any connected DataChannels every 5 seconds\n", d.Label(), d.ID())stat, err := os.Stat("/home/zhanglei/Downloads/《实践论》(原文)毛泽东.pdf")chunkSize = int(math.Ceil(float64(stat.Size() / int64(blockSize))))// 握手var chunk ChunkMessagechunk.Class = HANDSHAKEchunk.ChunkSize = uint64(chunkSize)handShakeBytes := serialize(&chunk)err = d.Send(handShakeBytes.Bytes())if err != nil {log.Error("%s", err)return}})// Register text message handlingd.OnMessage(func(msg webrtc.DataChannelMessage) {fmt.Printf("Message from DataChannel '%s': '%s'\n", d.Label(), string(msg.Data))data, err := unSerialize(msg.Data)if err != nil {log.Error("os.Open is : %s", err)return}if data.Class == ACK {handle, err := os.Open("/home/zhanglei/Downloads/《实践论》(原文)毛泽东.pdf")if err != nil {log.Error("os.Open is : %s", err)return}defer handle.Close()bufferBytes := make([]byte, blockSize)read, err := handle.Read(bufferBytes)if err != nil {log.Error("handle.Read is : %s", err)return}if read < blockSize {bufferBytes = bufferBytes[:read]}var chunk ChunkMessagechunk.Class = SENDchunk.ChunkSize = uint64(chunkSize)chunk.CurrentChunk = uint64(currentChunkSize)chunk.Data = bufferBytes// 打包发送err = d.Send(serialize(&chunk).Bytes())if err != nil {log.Error("%s", err)return}return}if data.Class == RECEIVE {handle, err := os.Open("/home/zhanglei/Downloads/《实践论》(原文)毛泽东.pdf")if err != nil {log.Error("os.Open is : %s", err)return}if data.CurrentChunk == uint64(chunkSize) {log.Info("data transfer finish")return}nextChunk := data.CurrentChunk + 1bytes := make([]byte, blockSize)read, err := handle.ReadAt(bytes, int64(nextChunk)*int64(blockSize))if err != nil {if !errors.Is(err, io.EOF) {log.Error("handle.Read is : %s", err)return}}if read < blockSize {bytes = bytes[:read]}var sendData ChunkMessagesendData.Class = SENDsendData.CurrentChunk = nextChunksendData.Data = bytessendData.ChunkSize = uint64(chunkSize)sendDataBytes := serialize(&sendData)log.Info(" read %d", nextChunk)err = d.Send(sendDataBytes.Bytes())if err != nil {log.Error("%s", err)return}// 最后一块if nextChunk == uint64(chunkSize) {d.Close()pc.Close()}return}})})_, err = pc.CreateDataChannel("sendDataChannel", nil)if err != nil {log.Error("error:%s", err)return nil}//channel.OnOpen(func() {////})//设置远端的sdpif err = pc.SetRemoteDescription(sdpDes); err != nil {log.Error("error:%s", err)return nil}//创建协商结果answer, err := pc.CreateAnswer(nil)if err != nil {log.Error("error:%s", err)return nil}pc.OnICECandidate(func(i *webrtc.ICECandidate) {fmt.Println("OnICECandidate")fmt.Println(i)})err = pc.SetLocalDescription(answer)if err != nil {log.Error("error:%s", err)return nil}//等待ice结束gatherCmp := webrtc.GatheringCompletePromise(pc)<-gatherCmp//将协商并且收集完candidate的answer，输出到控制台answerBytes, err := json.Marshal(*pc.LocalDescription())if err != nil {log.Error("error:%s", err)return nil}pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {fmt.Println("OnICECandidate")fmt.Println(candidate)})t := &Transfer{sdp:              sdp,pc:               pc,offset:           uint64(offset),currentChunkSize: currentChunkSize,chunkSize:        chunkSize,answerSdp:        answerBytes,blockSize:        blockSize,}return t

2、打包传输

传输没有采用protobuf，自己写了个二进制传输

func serialize(data *ChunkMessage) *bytes.Buffer {data.Version = CodeVersionwriteBuffer := bytes.NewBuffer(nil)writeBuffer.Write([]byte{data.Version})writeBuffer.Write([]byte{data.Class})// ChunkSizebinary.Write(writeBuffer, binary.BigEndian, data.ChunkSize)// CurrentChunkbinary.Write(writeBuffer, binary.BigEndian, data.CurrentChunk)// DataLendata.DataLen = uint64(len(data.Data))binary.Write(writeBuffer, binary.BigEndian, data.DataLen)if len(data.Data) > 0 {// 添加bodywriteBuffer.Write(data.Data)}return writeBuffer
}//判断我们系统中的字节序类型
func systemEdian() binary.ByteOrder {var i int = 0x1bs := (*[int(unsafe.Sizeof(0))]byte)(unsafe.Pointer(&i))if bs[0] == 0 {return binary.LittleEndian} else {return binary.BigEndian}
}func unSerialize(data []byte) (*ChunkMessage, error) {buf := bytes.NewBuffer(data)fmt.Println(buf)var chunk ChunkMessagebinary.Read(buf, systemEdian(), &chunk.Version)binary.Read(buf, systemEdian(), &chunk.Class)binary.Read(buf, systemEdian(), &chunk.ChunkSize)binary.Read(buf, systemEdian(), &chunk.CurrentChunk)binary.Read(buf, systemEdian(), &chunk.DataLen)//chunkSize := uint64(unsafe.Pointer(&buf.Bytes()))//chunk.ChunkSize = chunkSizereturn &chunk, nil
}

7、js前端webrtc提供报价

创建webrtc连接

var pcConfig = {'iceServers': [{'urls': 'stun:',}]};localConnection = new RTCPeerConnection(pcConfig);receiveDataChannel = localConnection.createDataChannel("receiveDataChannel")receiveDataChannel.binaryType = "arraybuffer"receiveDataChannel.addEventListener('open', dataChannel.onopen);receiveDataChannel.addEventListener('close', dataChannel.onclose);receiveDataChannel.addEventListener('message', dataChannel.onmessage);receiveDataChannel.addEventListener('error', dataChannel.onError);try {this.offer = await localConnection.createOffer();} catch (e) {console.log('Failed to create session description: ', e);return}try {await localConnection.setLocalDescription(this.offer)} catch (e) {console.log('Failed to create session description: ', e);return}//eyJ1c2VyX3V1aWQiOiI1ZmZkNDE0N2JkMTMyNWNmMjYwNDAyMWYwODA5OWUyMyIsImxvZ2luX3RpbWUiOjE2Njg0NzgzOTEsIm5vd190aW1lIjoxNjY5OTgzNzkwLCJyYW5fc3RyIjoiZDA3MTczNzI3NjFjMzY0MGU2NmRlYWI5YmYyODZhNzYiLCJzaWduIjoiZjc3NzI0YjZmMTc3MzczNmVhZWFkMTM2NzllNTE0NTcifQ==transfer.ws.send((JSON.stringify(downloadRequest)));

前端对收到的数据进行序列化和反序列化

function serialize(data) {var bufLen = protoColMinSize;if (!data.Data) {bufLen += 0;} else {bufLen += data.Data.length;}data.Version = 1;var protocolBuf = new ArrayBuffer(bufLen);const bufView = new DataView(protocolBuf);bufView.setUint8(0, data.Version);bufView.setUint8(1, data.Class);if (!data.ChunkSize) {data.ChunkSize = 0}bufView.setBigUint64(2, BigInt(data.ChunkSize));if (!data.CurrentChunk) {data.CurrentChunk = 0}bufView.setBigUint64(10, BigInt(data.CurrentChunk));if (data.Data && data.Data.length > 0) {bufView.setBigUint64(18, BigInt(data.Data.length));} else {bufView.setBigUint64(18, BigInt(0));}console.log(protocolBuf)return protocolBuf;
}function unSerialize(bytes) {var versionView = new DataView(bytes).getUint8(0);// 最小长度var classByteView = new DataView(bytes).getUint8(1);// chunk 长度var chunkSizeView = parseInt(new DataView(bytes).getBigUint64(2));var currentChunkView = parseInt(new DataView(bytes).getBigUint64(10));var bodyLenView = parseInt(new DataView(bytes).getBigUint64(18));var returnData = {Version: versionView,Class: classByteView,ChunkSize: (chunkSizeView),CurrentChunk: currentChunkView,PayloadLength: bodyLenView,Payload: [],};if (bodyLenView > 0) {returnData.Payload = new Uint8Array(bytes, protoColMinSize, bodyLenView)}return returnData;
}