normalize file names (#173)

server_multicast_handler.go

@@ -0,0 +1,108 @@
+package gortsplib
+
+import (
+	"fmt"
+	"net"
+
+	"github.com/aler9/gortsplib/v2/pkg/ringbuffer"
+)
+
+type typeAndPayload struct {
+	isRTP   bool
+	payload []byte
+}
+
+type serverMulticastHandler struct {
+	rtpl        *serverUDPListener
+	rtcpl       *serverUDPListener
+	writeBuffer *ringbuffer.RingBuffer
+
+	writerDone chan struct{}
+}
+
+func newServerMulticastHandler(s *Server) (*serverMulticastHandler, error) {
+	res := make(chan net.IP)
+	select {
+	case s.streamMulticastIP <- streamMulticastIPReq{res: res}:
+	case <-s.ctx.Done():
+		return nil, fmt.Errorf("terminated")
+	}
+	ip := <-res
+
+	rtpl, rtcpl, err := newServerUDPListenerMulticastPair(
+		s.ListenPacket,
+		s.WriteTimeout,
+		s.MulticastRTPPort,
+		s.MulticastRTCPPort,
+		ip,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	wb, _ := ringbuffer.New(uint64(s.WriteBufferCount))
+
+	h := &serverMulticastHandler{
+		rtpl:        rtpl,
+		rtcpl:       rtcpl,
+		writeBuffer: wb,
+		writerDone:  make(chan struct{}),
+	}
+
+	go h.runWriter()
+
+	return h, nil
+}
+
+func (h *serverMulticastHandler) close() {
+	h.rtpl.close()
+	h.rtcpl.close()
+	h.writeBuffer.Close()
+	<-h.writerDone
+}
+
+func (h *serverMulticastHandler) ip() net.IP {
+	return h.rtpl.ip()
+}
+
+func (h *serverMulticastHandler) runWriter() {
+	defer close(h.writerDone)
+
+	rtpAddr := &net.UDPAddr{
+		IP:   h.rtpl.ip(),
+		Port: h.rtpl.port(),
+	}
+
+	rtcpAddr := &net.UDPAddr{
+		IP:   h.rtcpl.ip(),
+		Port: h.rtcpl.port(),
+	}
+
+	for {
+		tmp, ok := h.writeBuffer.Pull()
+		if !ok {
+			return
+		}
+		data := tmp.(typeAndPayload)
+
+		if data.isRTP {
+			h.rtpl.write(data.payload, rtpAddr)
+		} else {
+			h.rtcpl.write(data.payload, rtcpAddr)
+		}
+	}
+}
+
+func (h *serverMulticastHandler) writePacketRTP(payload []byte) {
+	h.writeBuffer.Push(typeAndPayload{
+		isRTP:   true,
+		payload: payload,
+	})
+}
+
+func (h *serverMulticastHandler) writePacketRTCP(payload []byte) {
+	h.writeBuffer.Push(typeAndPayload{
+		isRTP:   false,
+		payload: payload,
+	})
+}