sophie/actioncam
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fixes #2, allow server goroutines to exit on deadlines and context cancel.

Browse source
This commit is contained in:
Jonas Köritz 2019-10-15 12:59:06 +02:00
parent 03ca0f77e6
commit e82f5a97c9
5 changed files with 108 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

102
libipcamera/RTPRelay.go
View file

@ -2,11 +2,13 @@ package libipcamera
import (
"bytes"
"context"
"encoding/binary"
"encoding/hex"
"io"
"log"
"net"
"time"
)
// RTPRelay holds information on the relaying stream listener
@ -15,11 +17,12 @@ type RTPRelay struct {
targetIP net.IP
targetPort int
listener net.PacketConn
context context.Context
}
// CreateRTPRelay creates a UDP listener that handles live data
// from the camera and forwards it as an RTP stream
func CreateRTPRelay(targetAddress net.IP, targetPort int) *RTPRelay {
func CreateRTPRelay(ctx context.Context, targetAddress net.IP, targetPort int) *RTPRelay {
conn, err := net.ListenPacket("udp", ":6669")
if err != nil {
@ -30,6 +33,7 @@ func CreateRTPRelay(targetAddress net.IP, targetPort int) *RTPRelay {
targetIP: targetAddress,
targetPort: targetPort,
listener: conn,
context: ctx,
}
if err != nil {
log.Printf("ERROR: %s\n", err)
@ -64,60 +68,70 @@ func handleCameraStream(relay RTPRelay, conn net.PacketConn) {
packetBuffer := bytes.Buffer{}
for {
if relay.close {
conn.SetReadDeadline(time.Now().Add(10 * time.Second))
select {
case <-relay.context.Done():
log.Println("Context Done")
rtpConn.Close()
relay.listener.Close()
break
}
conn.ReadFrom(buffer)
packetReader.Reset(buffer)
binary.Read(packetReader, binary.BigEndian, &header)
if header.Magic != 0xBCDE {
log.Printf("Received message with invalid magic (%x).", header.Magic)
break
}
if header.Length > 0 {
payload = make([]byte, header.Length)
_, err := io.ReadFull(packetReader, payload)
if err != nil {
log.Printf("Read Error: %s\n", err)
default:
if relay.close {
rtpConn.Close()
relay.listener.Close()
break
}
} else {
payload = []byte{}
}
switch header.MessageType {
case 0x0001: // H.264 Data
frameBuffer.Write(payload)
case 0x0002: // Time
// Append the Framebuffer
packetBuffer.Write(frameBuffer.Bytes())
conn.ReadFrom(buffer)
packetReader.Reset(buffer)
// Send out the packet
rtpConn.Write(packetBuffer.Bytes())
binary.Read(packetReader, binary.BigEndian, &header)
// Prepare the next packet
packetBuffer.Reset()
packetBuffer.Write([]byte{0x80, 0x63})
binary.Write(&packetBuffer, binary.BigEndian, sequenceNumber+1)
binary.Write(&packetBuffer, binary.BigEndian, (uint32)(elapsed)*90)
binary.Write(&packetBuffer, binary.BigEndian, (uint64(0)))
if header.Magic != 0xBCDE {
log.Printf("Received message with invalid magic (%x).", header.Magic)
break
}
// Reset the Framebuffer
frameBuffer.Reset()
sequenceNumber++
if header.Length > 0 {
payload = make([]byte, header.Length)
_, err := io.ReadFull(packetReader, payload)
if err != nil {
log.Printf("Read Error: %s\n", err)
break
}
} else {
payload = []byte{}
}
elapsed = binary.LittleEndian.Uint32(payload[12:])
default:
log.Printf("Received Unknown Message: %+v\n", header)
log.Printf("Payload:\n%s\n", hex.Dump(payload))
switch header.MessageType {
case 0x0001: // H.264 Data
frameBuffer.Write(payload)
case 0x0002: // Time
// Append the Framebuffer
packetBuffer.Write(frameBuffer.Bytes())
// Send out the packet
rtpConn.Write(packetBuffer.Bytes())
// Prepare the next packet
packetBuffer.Reset()
packetBuffer.Write([]byte{0x80, 0x63})
binary.Write(&packetBuffer, binary.BigEndian, sequenceNumber+1)
binary.Write(&packetBuffer, binary.BigEndian, (uint32)(elapsed)*90)
binary.Write(&packetBuffer, binary.BigEndian, (uint64(0)))
// Reset the Framebuffer
frameBuffer.Reset()
sequenceNumber++
elapsed = binary.LittleEndian.Uint32(payload[12:])
default:
log.Printf("Received Unknown Message: %+v\n", header)
log.Printf("Payload:\n%s\n", hex.Dump(payload))
}
}
}
rtpConn.Close()
}
// Stop stops listening for packets