mirrors_public/schollz_croc
0
0
Fork 0
mirror of https://github.com/schollz/croc.git synced 2025-10-11 13:21:00 +02:00
Code Activity

use detached channels

Browse source
This commit is contained in:
Zack Scholl 2019-04-05 13:15:06 -07:00
parent 800ee4a0d3
commit 3d9cac01e6
1 changed files with 115 additions and 79 deletions
Download patch file Download diff file Expand all files Collapse all files

66
src/croc/croc.go
View file

@ -568,8 +568,15 @@ func (c *Client) dataChannelReceive(num int) (err error) {
},
}
// Create a SettingEngine and enable Detach
s := webrtc.SettingEngine{}
s.DetachDataChannels()
// Create an API object with the engine
api := webrtc.NewAPI(webrtc.WithSettingEngine(s))
// Create a new RTCPeerConnection
c.peerConnection[num], err = webrtc.NewPeerConnection(config)
c.peerConnection[num], err = api.NewPeerConnection(config)
if err != nil {
return
}
@ -586,16 +593,26 @@ func (c *Client) dataChannelReceive(num int) (err error) {
// Register channel opening handling
d.OnOpen(func() {
c.log.Debugf("Data channel '%s'-'%d' open. Random messages will now be sent to any connected DataChannels every 5 seconds\n", d.Label(), d.ID())
})
// Detach the data channel
raw, dErr := d.Detach()
if dErr != nil {
panic(dErr)
}
startTime := false
timer := time.Now()
piecesToDo := make(map[int64]bool)
for i := int64(0); i < c.FilesToTransfer[c.FilesToTransferCurrentNum].Size; i += BufferSize {
piecesToDo[i] = true
// Handle reading from the data channel
go func(d io.Reader) {
for {
buffer := make([]byte, BufferSize*2)
n, err := d.Read(buffer)
if err != nil {
fmt.Println("Datachannel closed; Exit the readloop:", err)
return
}
d.OnMessage(func(msg webrtc.DataChannelMessage) {
if bytes.Equal([]byte("done"), msg.Data) {
if bytes.Equal([]byte("done"), buffer[:n]) {
c.log.Debug(time.Since(timer))
c.log.Debug("telling transfer is over")
err = c.redisdb.Publish(c.nameOutChannel, Message{
@ -613,19 +630,23 @@ func (c *Client) dataChannelReceive(num int) (err error) {
timer = time.Now()
}
var chunk Chunk
errM := json.Unmarshal(msg.Data, &chunk)
errM := json.Unmarshal(buffer[:n], &chunk)
if errM != nil {
panic(errM)
}
var n int
var nBytes int
c.mutex.Lock()
n, err = c.CurrentFile.WriteAt(chunk.Bytes, chunk.Location)
nBytes, err = c.CurrentFile.WriteAt(chunk.Bytes, chunk.Location)
// c.log.Debugf("wrote %d bytes to %d (%d)", n, chunk.Location,num)
c.mutex.Unlock()
if err != nil {
panic(err)
}
c.bar.Add(n)
c.bar.Add(nBytes)
}
}(raw)
})
})
@ -646,8 +667,15 @@ func (c *Client) dataChannelSend(num int) (err error) {
},
}
// Create a SettingEngine and enable Detach
s := webrtc.SettingEngine{}
s.DetachDataChannels()
// Create an API object with the engine
api := webrtc.NewAPI(webrtc.WithSettingEngine(s))
// Create a new RTCPeerConnection
c.peerConnection[num], err = webrtc.NewPeerConnection(config)
c.peerConnection[num], err = api.NewPeerConnection(config)
if err != nil {
return
}
@ -666,8 +694,13 @@ func (c *Client) dataChannelSend(num int) (err error) {
c.dataChannel[num].OnOpen(func() {
fmt.Printf("Data channel '%s'-'%d' open. Random messages will now be sent to any connected DataChannels every 5 seconds\n", c.dataChannel[num].Label(), c.dataChannel[num].ID())
time.Sleep(100 * time.Microsecond)
// Detach the data channel
raw, dErr := c.dataChannel[num].Detach()
if dErr != nil {
panic(dErr)
}
go func(d io.Writer) {
pathToFile := path.Join(c.FilesToTransfer[c.FilesToTransferCurrentNum].FolderSource, c.FilesToTransfer[c.FilesToTransferCurrentNum].Name)
c.log.Debugf("sending '%s'", pathToFile)
@ -697,7 +730,7 @@ func (c *Client) dataChannelSend(num int) (err error) {
}
dataToSend, _ := json.Marshal(mSend)
c.bar.Add(bytesread)
err = c.dataChannel[num].Send(dataToSend)
_, err = d.Write(dataToSend)
if err != nil {
c.log.Debug("Could not send on data channel", err.Error())
continue
@ -709,11 +742,14 @@ func (c *Client) dataChannelSend(num int) (err error) {
chunkNum += 1.0
}
c.log.Debug("sending done signal")
err = c.dataChannel[num].Send([]byte("done"))
_, err = d.Write([]byte("done"))
if err != nil {
c.log.Debug(err)
}
time.Sleep(1 * time.Second)
}(raw)
})
// Register text message handling