Refactor message processor to better reflect the actual "Valeria" API

main.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	stdlog "log"
 	"os"
-	"os/signal"
 	"strings"
 	"time"
 
@@ -32,6 +31,7 @@ Usage:
   qvh gstreamer [--pipeline=<pipeline>] [--examples] [--udid=<udid>] [-v]
   qvh diagnostics <outfile> [--dump=<dumpfile>] [--udid=<udid>]
   qvh --version | version
+  qvh test
 
 
 Options:
@@ -88,13 +88,14 @@ The commands work as following:
 	if err != nil {
 		printErrJSON(err, "no device found to use")
 	}
-	checkDeviceIsPaired(device)
+	//checkDeviceIsPaired(device)
 
 	activateCommand, _ := arguments.Bool("activate")
 	if activateCommand {
 		activate(device)
 		return
 	}
+
 	audioCommand, _ := arguments.Bool("audio")
 	if audioCommand {
 		outfile, err := arguments.String("<outfile>")
@@ -226,7 +227,7 @@ func recordAudioGst(outfile string, device screencapture.IosDevice, audiotype st
 		printErrJSON(err, "Failed creating custom pipeline")
 		return
 	}
-	startWithConsumer(gStreamer, device, true)
+	screencapture.StartWithConsumer(gStreamer, device, true)
 }
 
 func runDiagnostics(outfile string, dump bool, dumpFile string, device screencapture.IosDevice) {
@@ -238,10 +239,10 @@ func runDiagnostics(outfile string, dump bool, dumpFile string, device screencap
 	defer metricsFile.Close()
 	consumer := diagnostics.NewDiagnosticsConsumer(metricsFile, time.Second*10)
 	if dump {
-		startWithConsumerDump(consumer, device, dumpFile)
+		screencapture.StartWithConsumerDump(consumer, device, dumpFile)
 		return
 	}
-	startWithConsumer(consumer, device, false)
+	screencapture.StartWithConsumer(consumer, device, false)
 }
 
 func recordAudioWav(outfile string, device screencapture.IosDevice) {
@@ -268,7 +269,7 @@ func recordAudioWav(outfile string, device screencapture.IosDevice) {
 		}
 
 	}()
-	startWithConsumer(wavFileWriter, device, true)
+	screencapture.StartWithConsumer(wavFileWriter, device, true)
 }
 
 func startGStreamerWithCustomPipeline(device screencapture.IosDevice, pipelineString string) {
@@ -278,13 +279,13 @@ func startGStreamerWithCustomPipeline(device screencapture.IosDevice, pipelineSt
 		printErrJSON(err, "Failed creating custom pipeline")
 		return
 	}
-	startWithConsumer(gStreamer, device, false)
+	screencapture.StartWithConsumer(gStreamer, device, false)
 }
 
 func startGStreamer(device screencapture.IosDevice) {
 	log.Debug("Starting Gstreamer")
 	gStreamer := gstadapter.New()
-	startWithConsumer(gStreamer, device, false)
+	screencapture.StartWithConsumer(gStreamer, device, false)
 }
 
 // Just dump a list of what was discovered to the console
@@ -353,72 +354,7 @@ func record(h264FilePath string, wavFilePath string, device screencapture.IosDev
 		}
 
 	}()
-	startWithConsumer(writer, device, false)
-}
-
-func startWithConsumer(consumer screencapture.CmSampleBufConsumer, device screencapture.IosDevice, audioOnly bool) {
-	var err error
-	device, err = screencapture.EnableQTConfig(device)
-	if err != nil {
-		printErrJSON(err, "Error enabling QT config")
-		return
-	}
-
-	adapter := screencapture.UsbAdapter{}
-	stopSignal := make(chan interface{})
-	waitForSigInt(stopSignal)
-
-	mp := screencapture.NewMessageProcessor(&adapter, stopSignal, consumer, audioOnly)
-
-	err = adapter.StartReading(device, &mp, stopSignal)
-	consumer.Stop()
-	if err != nil {
-		printErrJSON(err, "failed connecting to usb")
-	}
-}
-
-func startWithConsumerDump(consumer screencapture.CmSampleBufConsumer, device screencapture.IosDevice, dumpPath string) {
-	var err error
-	device, err = screencapture.EnableQTConfig(device)
-	if err != nil {
-		printErrJSON(err, "Error enabling QT config")
-		return
-	}
-
-	inboundMessagesFile, err := os.Create("inbound-" + dumpPath)
-	if err != nil {
-		log.Fatalf("Could not open file: %v", err)
-	}
-	defer inboundMessagesFile.Close()
-	outboundMessagesFile, err := os.Create("outbound-" + dumpPath)
-	if err != nil {
-		log.Fatalf("Could not open file: %v", err)
-	}
-	defer outboundMessagesFile.Close()
-	log.Debug("Start dumping all binary transfer")
-	adapter := screencapture.UsbAdapter{Dump: true, DumpInWriter: inboundMessagesFile, DumpOutWriter: outboundMessagesFile}
-	stopSignal := make(chan interface{})
-	waitForSigInt(stopSignal)
-
-	mp := screencapture.NewMessageProcessor(&adapter, stopSignal, consumer, false)
-
-	err = adapter.StartReading(device, &mp, stopSignal)
-	consumer.Stop()
-	if err != nil {
-		printErrJSON(err, "failed connecting to usb")
-	}
-}
-
-func waitForSigInt(stopSignalChannel chan interface{}) {
-	c := make(chan os.Signal, 1)
-	signal.Notify(c, os.Interrupt)
-	go func() {
-		for sig := range c {
-			log.Debugf("Signal received: %s", sig)
-			var stopSignal interface{}
-			stopSignalChannel <- stopSignal
-		}
-	}()
+	screencapture.StartWithConsumer(writer, device, false)
 }
 
 func checkDeviceIsPaired(device screencapture.IosDevice) {

screencapture/interfaces.go

@@ -18,3 +18,9 @@ type UsbDataReceiver interface {
 type UsbWriter interface {
 	WriteDataToUsb(data []byte)
 }
+
+//UsbWriter can be used to send data to a USB Endpoint
+type UsbWriterNew interface {
+	WriteDataToUsb(data []byte) error
+	ReadFrame() ([]byte, error)
+}

screencapture/messageprocessor.go

@@ -83,7 +83,7 @@ func (mp *MessageProcessor) handleSyncPacket(data []byte) {
 		}
 		log.Debugf("Rcv:%s", ogPacket.String())
 
-		replyBytes := ogPacket.NewReply()
+		replyBytes := ogPacket.NewReply(0)
 		log.Debugf("Send OG-REPLY {correlation:%x}", ogPacket.CorrelationID)
 		mp.usbWriter.WriteDataToUsb(replyBytes)
 	case packet.CWPA:

screencapture/messageprocessor_test.go

@@ -83,9 +83,9 @@ func TestMessageProcessorRespondsCorrectlyToSyncMessages(t *testing.T) {
 			description: "Expect correct reply for cwpa",
 		},
 		{
-			receivedData:  loadFromFile("og-request")[4:],
-			expectedReply: [][]byte{loadFromFile("og-reply")},
-			description:   "Expect correct reply for og",
+			receivedData:  loadFromFile("gocmd-request")[4:],
+			expectedReply: [][]byte{loadFromFile("gocmd-reply")},
+			description:   "Expect correct reply for gocmd",
 		},
 		{
 			receivedData:  loadFromFile("stop-request")[4:],

screencapture/packet/sync_og.go

@@ -25,12 +25,12 @@ func NewSyncOgPacketFromBytes(data []byte) (SyncOgPacket, error) {
 }
 
 //NewReply returns a []byte containing the default reply for a SyncOgPacket
-func (sp SyncOgPacket) NewReply() []byte {
+func (sp SyncOgPacket) NewReply(response uint64) []byte {
 	responseBytes := make([]byte, 24)
 	binary.LittleEndian.PutUint32(responseBytes, 24)
 	binary.LittleEndian.PutUint32(responseBytes[4:], ReplyPacketMagic)
 	binary.LittleEndian.PutUint64(responseBytes[8:], sp.CorrelationID)
-	binary.LittleEndian.PutUint64(responseBytes[16:], 0)
+	binary.LittleEndian.PutUint64(responseBytes[16:], response)
 
 	return responseBytes
 

screencapture/usbadapter-new.go

@@ -0,0 +1,163 @@
+package screencapture
+
+import (
+	"encoding/binary"
+	"fmt"
+	"io"
+
+	"github.com/pkg/errors"
+
+	"github.com/google/gousb"
+	log "github.com/sirupsen/logrus"
+)
+
+//UsbAdapterNew reads and writes from AV Quicktime USB Bulk endpoints
+type UsbAdapterNew struct {
+	outEndpoint   *gousb.OutEndpoint
+	Dump          bool
+	DumpOutWriter io.Writer
+	DumpInWriter  io.Writer
+	stream        *gousb.ReadStream
+	usbDevice     *gousb.Device
+	contextClose  func()
+	iface         *gousb.Interface
+	iosDevice     IosDevice
+}
+
+//WriteDataToUsb implements the UsbWriter interface and sends the byte array to the usb bulk endpoint.
+func (usbAdapter *UsbAdapterNew) WriteDataToUsb(bytes []byte) error {
+	_, err := usbAdapter.outEndpoint.Write(bytes)
+	if err != nil {
+		return err
+	}
+	if usbAdapter.Dump {
+		_, err := usbAdapter.DumpOutWriter.Write(bytes)
+		if err != nil {
+			log.Fatalf("Failed dumping data:%v", err)
+		}
+	}
+	return nil
+}
+
+func (usbAdapter *UsbAdapterNew) InitializeUSB(device IosDevice) error {
+	ctx, cleanUp := createContext()
+	usbAdapter.contextClose = cleanUp
+	usbAdapter.outEndpoint = &gousb.OutEndpoint{}
+	usbAdapter.stream = &gousb.ReadStream{}
+	usbAdapter.usbDevice = &gousb.Device{}
+	usbAdapter.iface = &gousb.Interface{}
+	usbAdapter.iosDevice = device
+
+	usbDevice, err := OpenDevice(ctx, device)
+	if err != nil {
+		return err
+	}
+	if !device.IsActivated() {
+		return errors.New("device not activated for screen mirroring")
+	}
+
+	confignum, _ := usbDevice.ActiveConfigNum()
+	log.Debugf("Config is active: %d, QT config is: %d", confignum, device.QTConfigIndex)
+
+	config, err := usbDevice.Config(device.QTConfigIndex)
+	if err != nil {
+		return errors.New("Could not retrieve config")
+	}
+
+	log.Debugf("QT Config is active: %s", config.String())
+
+	iface, err := findAndClaimQuickTimeInterface(config)
+	if err != nil {
+		log.Debug("could not get Quicktime Interface")
+		return err
+	}
+	log.Debugf("Got QT iface:%s", iface.String())
+
+	inboundBulkEndpointIndex, inboundBulkEndpointAddress, err := findBulkEndpoint(iface.Setting, gousb.EndpointDirectionIn)
+	if err != nil {
+		return err
+	}
+
+	outboundBulkEndpointIndex, outboundBulkEndpointAddress, err := findBulkEndpoint(iface.Setting, gousb.EndpointDirectionOut)
+	if err != nil {
+		return err
+	}
+
+	err = clearFeature(usbDevice, inboundBulkEndpointAddress, outboundBulkEndpointAddress)
+	if err != nil {
+		return err
+	}
+
+	inEndpoint, err := iface.InEndpoint(inboundBulkEndpointIndex)
+	if err != nil {
+		log.Error("couldnt get InEndpoint")
+		return err
+	}
+	log.Debugf("Inbound Bulk: %s", inEndpoint.String())
+
+	outEndpoint, err := iface.OutEndpoint(outboundBulkEndpointIndex)
+	if err != nil {
+		log.Error("couldnt get OutEndpoint")
+		return err
+	}
+
+	stream, err := inEndpoint.NewStream(4096, 5)
+	if err != nil {
+		log.Fatal("couldnt create stream")
+		return err
+	}
+	log.Debug("Endpoint claimed")
+	log.Debugf("Outbound Bulk: %s", outEndpoint.String())
+	usbAdapter.outEndpoint = outEndpoint
+	usbAdapter.stream = stream
+	usbAdapter.usbDevice = usbDevice
+	usbAdapter.iface = iface
+
+	return nil
+}
+
+func (usbAdapter *UsbAdapterNew) Close() error {
+	log.Info("Closing usb stream")
+
+	err := usbAdapter.stream.Close()
+	if err != nil {
+		log.Error("Error closing stream", err)
+	}
+	log.Info("Closing usb interface")
+	usbAdapter.iface.Close()
+
+	sendQTDisableConfigControlRequest(usbAdapter.usbDevice)
+	log.Debug("Resetting device config")
+	_, err = usbAdapter.usbDevice.Config(usbAdapter.iosDevice.UsbMuxConfigIndex)
+	if err != nil {
+		log.Warn(err)
+	}
+	usbAdapter.contextClose()
+	return nil
+}
+
+func (usbAdapter *UsbAdapterNew) ReadFrame() ([]byte, error) {
+	lengthBuffer := make([]byte, 4)
+	for {
+		n, err := io.ReadFull(usbAdapter.stream, lengthBuffer)
+		if err != nil {
+			return []byte{}, fmt.Errorf("failed reading 4bytes length with err:%s only received: %d", err, n)
+		}
+		//the 4 bytes header are included in the length, so we need to subtract them
+		//here to know how long the payload will be
+		length := binary.LittleEndian.Uint32(lengthBuffer) - 4
+		dataBuffer := make([]byte, length)
+
+		n, err = io.ReadFull(usbAdapter.stream, dataBuffer)
+		if err != nil {
+			return []byte{}, err
+		}
+		if usbAdapter.Dump {
+			_, err := usbAdapter.DumpInWriter.Write(dataBuffer)
+			if err != nil {
+				log.Fatalf("Failed dumping data:%v", err)
+			}
+		}
+		return dataBuffer, nil
+	}
+}