Passing all test cases! Just need to see if we can actually get something watchable

packets/MpegTs.go

@@ -28,7 +28,7 @@ LICENSE
 
 package packets
 
-import "fmt"
+import _"fmt"
 
 // Length of some fields in bits
 const (
@@ -81,15 +81,9 @@ func (p *MpegTsPacket) ToByteSlice() (output []byte) {
 	for ii := 4; ii-4 < len(p.AF); ii++ {
 		output[ii] = p.AF[ii-4]
 	}
-	//copy(output[4:4+len(p.AF)],p.AF)
-	//headerSize := packetLength-len(p.Payload)
-	fmt.Printf("Length of AF: %v\n", len(p.AF))
 	payloadIndex := 4 + len(p.AF)
 	for ii := payloadIndex; ii < packetLength; ii++ {
 		output[ii] = p.Payload[ii-payloadIndex]
-		fmt.Println(ii)
 	}
-	fmt.Printf("Packet size: %v\n",len(output))
-	//fmt.Printf("Packet: %v\n",output)
 	return
 }

packets/RtpToTsConverter.go

@@ -35,6 +35,8 @@ type RtpToTsConverter interface {
 type rtpToTsConverter struct {
   TsChan <-chan *MpegTsPacket
   tsChan chan<- *MpegTsPacket
+  InputChan chan<- *RtpPacket
+  inputChan <-chan *RtpPacket
   currentTsPacket *MpegTsPacket
   payloadByteChan chan byte
   currentCC byte
@@ -45,34 +47,36 @@ func NewRtpToTsConverter() (c *rtpToTsConverter) {
   tsChan := make(chan *MpegTsPacket,100)
   c.TsChan = tsChan
   c.tsChan = tsChan
-  c.payloadByteChan = make(chan byte, 10000)
+  inputChan := make(chan *RtpPacket,100)
+  c.InputChan = inputChan
+  c.inputChan = inputChan
   c.currentCC = 0
   return
 }
 
-func (c* rtpToTsConverter) Convert(rtpSession *Session) {
+func (c* rtpToTsConverter) Convert() {
   nalUnitChan := make(chan NALUnit, 1000)
+  pesPktChan := make(chan []byte, 1000)
+  pesDataChan := make(chan byte, 1000)
+  payloadByteChan := make(chan byte, 10000)
   // Get nal units from incoming rtp
-  go func(){
   for {
-      rtpPacket := <-rtpSession.RtpChan
+    select {
+    default:
+    case rtpPacket := <-c.inputChan:
       if GetNalType( rtpPacket.Payload ) == 28 {
         nalUnitChan<-ParseNALFragment(rtpPacket.Payload)
       } else {
-      //  nalUnitChan<-ParseSPSPPS(rtpPacket.Payload)
+        nalUnitChan<-ParseNALSpsPps(rtpPacket.Payload)
       }
+    case nalUnit := <-nalUnitChan:
+      nalUnitByteSlice := nalUnit.ToByteSlice()
+      for ii := range nalUnitByteSlice {
+        pesDataChan<-nalUnitByteSlice[ii]
       }
-  }()
-  // Create PES packets from NAl units
-  pesPktChan := make(chan []byte, 100)
-  go func(){
-    pesDataChan := make(chan byte, 100)
-    for {
-      nalUnit:=<-nalUnitChan
       if nalFragment, isNALFragment := nalUnit.(*NALFragment);
-          (isNALFragment && nalFragment.Start) || !isNALFragment {
+          (isNALFragment && nalFragment.End) || !isNALFragment {
         pesDataChanLen := len(pesDataChan)
-        if pesDataChanLen > 0 {
         pesPkt := new(PESPacket)
         pesPkt.StreamID = 0xE0
         pesPkt.Length = uint16( 3 + pesDataChanLen )
@@ -95,22 +99,14 @@ func (c* rtpToTsConverter) Convert(rtpSession *Session) {
         }
         pesPktChan<-pesPkt.ToByteSlice()
       }
-      }
+
-      nalUnitByteSlice := nalUnit.ToByteSlice()
+    case pesPkt := <-pesPktChan:
-      for ii := range nalUnitByteSlice {
-        pesDataChan<-nalUnitByteSlice[ii]
-      }
-    }
-  }()
-  // Create mpegts using pes packets
-  for {
-    pesPkt := <-pesPktChan
       for ii:=range pesPkt {
-      c.payloadByteChan<-pesPkt[ii]
+        payloadByteChan<-pesPkt[ii]
       }
       firstPacket:=true
-    for len(c.payloadByteChan) > 0 {
+      for len(payloadByteChan) > 0 {
-      lengthOfByteChan := len(c.payloadByteChan)
+        lengthOfByteChan := len(payloadByteChan)
         c.currentTsPacket = new(MpegTsPacket)
         c.currentTsPacket.SyncByte = 0x47
         c.currentTsPacket.TEI = false
@@ -134,16 +130,17 @@ func (c* rtpToTsConverter) Convert(rtpSession *Session) {
         c.currentTsPacket.AF[0] = byte(1 + stuffingLength)
         c.currentTsPacket.AF[1] = 0
         if c.currentTsPacket.PUSI {
-        c.currentTsPacket.AF[1] = 0x40
+          c.currentTsPacket.AF[1] = 0x00
         }
         for ii := 0; ii < stuffingLength; ii++ {
           c.currentTsPacket.AF[2+ii] = 0xFF
         }
         c.currentTsPacket.Payload = make([]byte, payloadLength)
         for ii:=0; ii < payloadLength; ii++ {
-        c.currentTsPacket.Payload[ii] = <-c.payloadByteChan
+          c.currentTsPacket.Payload[ii] = <-payloadByteChan
         }
         c.tsChan<-c.currentTsPacket
       }
     }
   }
+}

packets/packets_test.go

@@ -31,14 +31,15 @@ package packets
 import (
 	//"bytes"
 	"fmt"
-	"io"
+	_"io"
-	"log"
+	_"log"
-	"net"
+	_"net"
 	"reflect"
 	"testing"
 	"time"
+	"math/rand"
 
-	"github.com/beatgammit/rtsp"
+	_"github.com/beatgammit/rtsp"
 )
 
 /*******************************************************
@@ -53,12 +54,14 @@ const (
 
 /* Let's see if we can connect to an rtsp device then read an rtp stream,
 and then convert the rtp packets to mpegts packets and output. */
+/*
 func TestRTSP(t *testing.T) {
 	sess := rtsp.NewSession()
 	res, err := sess.Options(rtspUrl)
 	if err != nil {
 		t.Errorf("Shouldn't have got error: %v\n", err)
 	}
+
 	res, err = sess.Describe(rtspUrl)
 	if err != nil {
 		log.Fatalln(err)
@@ -143,6 +146,7 @@ func TestRTP(t *testing.T) {
 		fmt.Printf("RTP packet: %v\n", rtpPacket)
 	}
 }
+*/
 
 /*******************************************************
 Testing stuff related to the Nal.go file
@@ -352,7 +356,160 @@ func TestMpegTsToByteSlice(t *testing.T){
 /*******************************************************
 RtpToTsConverter testing
 ********************************************************/
-
 func TestRtpToTsConverter(t *testing.T){
-
+	converter := NewRtpToTsConverter()
+	go converter.Convert()
+	// Create first rtp packet
+	rtpPacket1 := new(RtpPacket)
+	rtpPacket1.Version = 2
+	rtpPacket1.Padding = false
+	rtpPacket1.Ext = false
+	rtpPacket1.CC = 0
+	rtpPacket1.Marker = true
+	rtpPacket1.PayloadType = 0xE3
+	rtpPacket1.SequenceNumber = 1
+	rtpPacket1.Timestamp = 200
+	rtpPacket1.SyncSource = 0
+	rtpPacket1.CSRC = nil
+	rtpPacket1.ExtHeader = 0
+	rtpPacket1.ExtData = nil
+	nalFragment := new(NALFragment)
+	nalFragment.ThreeNUBs = 0x02
+	nalFragment.FragmentType = byte(28)
+	nalFragment.Start = true
+	nalFragment.End = false
+	nalFragment.Reserved = true
+	nalFragment.FiveNUBs = 0x03
+	nalFragment.Data = make([]byte,98)
+	rand.Seed(int64(time.Now().Nanosecond()))
+	for i := range nalFragment.Data {
+		nalFragment.Data[i] = byte(rand.Intn(255))
+		fmt.Printf(" %v ", nalFragment.Data[i])
+	}
+	rtpPacket1.Payload = make([]byte,100)
+	copy(rtpPacket1.Payload[:], nalFragment.ToByteSlice())
+	fmt.Println(rtpPacket1.Payload)
+	converter.InputChan<-rtpPacket1
+	// Create second rtp packet
+	rtpPacket2 := new(RtpPacket)
+	rtpPacket2.Version = 2
+	rtpPacket2.Padding = false
+	rtpPacket2.Ext = false
+	rtpPacket2.CC = 0
+	rtpPacket2.Marker = false
+	rtpPacket2.PayloadType = 0xE3
+	rtpPacket2.SequenceNumber = 2
+	rtpPacket2.Timestamp = 300
+	rtpPacket2.SyncSource = 0
+	rtpPacket2.CSRC = nil
+	rtpPacket2.ExtHeader = 0
+	rtpPacket2.ExtData = nil
+	nalFragment = new(NALFragment)
+	nalFragment.ThreeNUBs = 0x02
+	nalFragment.FragmentType = byte(28)
+	nalFragment.Start = false
+	nalFragment.End = true
+	nalFragment.Reserved = true
+	nalFragment.FiveNUBs = 0x03
+	nalFragment.Data = make([]byte,198)
+	for i := range nalFragment.Data {
+		nalFragment.Data[i] = byte(rand.Intn(255))
+	}
+	rtpPacket2.Payload = make([]byte,200)
+	copy(rtpPacket2.Payload[:], nalFragment.ToByteSlice())
+	converter.InputChan<-rtpPacket2
+
+	// Create first expected tsPacket
+	afField := make([]byte, 2)
+	afField[0] = byte(1)
+	afField[1] = byte(0)
+	pesPkt := new(PESPacket)
+	pesPkt.StreamID = 0xE0
+	pesPkt.Length = uint16( 3 + 300 )
+	pesPkt.ScramblingControl = 0
+	pesPkt.Priority = true
+	pesPkt.DAI = false
+	pesPkt.Copyright = false
+	pesPkt.Original = true
+	pesPkt.PDI = 0
+	pesPkt.ESCR = false
+	pesPkt.ESRate = false
+	pesPkt.DSMTrickMode = false
+	pesPkt.ACI = false
+	pesPkt.CRC = false
+	pesPkt.Ext = false
+	pesPkt.HeaderLength = 0
+	pesPkt.Data = make([]byte,300)
+	for ii:=0; ii<100; ii++ {
+		pesPkt.Data[ii] = rtpPacket1.Payload[ii]
+	}
+	for ii:=100; ii <300; ii++ {
+		pesPkt.Data[ii] = rtpPacket2.Payload[ii-100]
+	}
+
+	pesPacketAsByteSlice := pesPkt.ToByteSlice()
+
+	data := make([]byte, 182)
+
+	copy(data[:],pesPacketAsByteSlice[:182])
+
+	expectedPkt1 := MpegTsPacket{
+		byte(0x47), 	// sync byte
+		bool(false),	// TEI
+		bool(true),	// PUSI
+		bool(false),	// Priority
+		uint16(256),	// PID
+		byte(0),			// TSC
+		byte(3),			// AFC
+		byte(0), 			// CC
+		afField,			// AF
+		data,
+	}
+
+	data = make([]byte, len(pesPacketAsByteSlice)-182)
+
+	copy(data[:], pesPacketAsByteSlice[182:])
+
+	afField = make([]byte, 2+(182-(len(pesPacketAsByteSlice)-182)))
+	afField[0] = byte(1+(182-(len(pesPacketAsByteSlice)-182)))
+	afField[1] = byte(0)
+	for ii := 2; ii < len(afField); ii++ {
+		afField[ii] = 0xFF
+	}
+
+	expectedPkt2 := MpegTsPacket{
+		byte(0x47), 	// sync byte
+		bool(false),	// TEI
+		bool(false),	// PUSI
+		bool(false),	// Priority
+		uint16(256),	// PID
+		byte(0),			// TSC
+		byte(3),			// AFC
+		byte(1), 			// CC
+		afField,			// AF
+		data,
+	}
+	// Now let's get our two Ts packets from the converter and see if they're  G
+	tsPacket := <-converter.TsChan
+	expectedPkt1AsByteSlice := expectedPkt1.ToByteSlice()
+	tsPacketAsByteSlice := tsPacket.ToByteSlice()
+	for ii := range expectedPkt1AsByteSlice {
+		if expectedPkt1AsByteSlice[ii] != tsPacketAsByteSlice[ii] {
+			t.Errorf("Not equal! Byte: %v Exptected: %v Got: %v\n",ii,
+				expectedPkt1AsByteSlice[ii],tsPacketAsByteSlice[ii])
+		}
+	}
+	fmt.Printf("Expected packet: %v\n", expectedPkt1.ToByteSlice())
+	fmt.Printf("Got packet: %v\n", tsPacket.ToByteSlice())
+	tsPacket = <-converter.TsChan
+	expectedPkt2AsByteSlice := expectedPkt2.ToByteSlice()
+	tsPacketAsByteSlice = tsPacket.ToByteSlice()
+	for ii := range expectedPkt2AsByteSlice {
+		if expectedPkt2AsByteSlice[ii] != tsPacketAsByteSlice[ii] {
+			t.Errorf("Not equal! Byte: %v Exptected: %v Got: %v\n",ii,
+				expectedPkt2AsByteSlice[ii],tsPacketAsByteSlice[ii])
+		}
+	}
+	fmt.Printf("Expected packet: %v\n", expectedPkt2.ToByteSlice())
+	fmt.Printf("Got packet: %v\n", tsPacket.ToByteSlice())
 }

revid/revidv2.go

@@ -286,6 +286,11 @@ func input(input string, output string) {
 	}
 	rtpSession := packets.NewSession(rtpConn,rtcpConn)
 	converter := packets.NewRtpToTsConverter()
+	go func(){
+		for{
+			converter.InputChan<-<-rtpSession.RtpChan
+		}
+	}
 	go converter.Convert(rtpSession)
 	clipSize := 0
 	packetCount := 0