First implementation

part/record.go

@@ -0,0 +1,145 @@
+package part
+
+import (
+	"github.com/cyrilix/robocar-base/service"
+	"github.com/cyrilix/robocar-protobuf/go/events"
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/golang/protobuf/proto"
+	log "github.com/sirupsen/logrus"
+	"sync"
+)
+
+func NewRecorder(client mqtt.Client, recordTopic, cameraTopic, steeringTopic, switchRecordTopic string) *Recorder {
+	return &Recorder{
+		client:        client,
+		recordTopic:   recordTopic,
+		cameraTopic:   cameraTopic,
+		steeringTopic: steeringTopic,
+		switchRecordTopic: switchRecordTopic,
+		enabled:       false,
+		cancel:        make(chan interface{}),
+	}
+}
+
+type Recorder struct {
+	client                     mqtt.Client
+	recordTopic                string
+	cameraTopic, steeringTopic, switchRecordTopic string
+
+	muSteeringMsg   sync.Mutex
+	currentSteering *events.SteeringMessage
+
+	muEnabled sync.RWMutex
+	enabled   bool
+
+	cancel chan interface {
+	}
+}
+
+func (r *Recorder) Start() error {
+	registerCallBacks(r)
+
+	for {
+		select {
+		case <-r.cancel:
+			log.Infof("Stop service")
+			return nil
+		}
+	}
+}
+
+func (r *Recorder) Stop() {
+	close(r.cancel)
+	service.StopService("record", r.client, r.cameraTopic, r.steeringTopic)
+}
+
+func (r *Recorder) onSwitchRecord(_ mqtt.Client, message mqtt.Message) {
+	var msg events.SwitchRecordMessage
+	err := proto.Unmarshal(message.Payload(), &msg)
+	if err != nil {
+		log.Errorf("unable to unmarshal protobuf %T: %v", msg, err)
+		return
+	}
+
+	r.muEnabled.Lock()
+	defer r.muEnabled.Unlock()
+	r.enabled = msg.GetEnabled()
+}
+
+func (r *Recorder) onSteering(_ mqtt.Client, message mqtt.Message) {
+	var msg events.SteeringMessage
+	err := proto.Unmarshal(message.Payload(), &msg)
+	if err != nil {
+		log.Errorf("unable to unmarshal protobuf %T: %v", msg, err)
+		return
+	}
+
+	r.muSteeringMsg.Lock()
+	defer r.muSteeringMsg.Unlock()
+	r.currentSteering = &msg
+}
+
+func (r *Recorder) onFrame(_ mqtt.Client, message mqtt.Message) {
+	if !r.Enabled() {
+		return
+	}
+
+	var msg events.FrameMessage
+	err := proto.Unmarshal(message.Payload(), &msg)
+	if err != nil {
+		log.Errorf("unable to unmarshal protobuf FrameMessage: %v", err)
+		return
+	}
+
+	steering := r.CurrentSteering()
+	if steering == nil {
+		log.Warningf("no current steeringMsg, skip frameMsg %v", msg.GetId().Id)
+		return
+	}
+
+	record := events.RecordMessage{
+		Frame:    &msg,
+		Steering: steering,
+	}
+
+	payload, err := proto.Marshal(&record)
+	if err != nil {
+		log.Errorf("unable to marshal message %v: %v", record, err)
+		return
+	}
+	publish(r.client, r.recordTopic, &payload)
+}
+
+var publish = func(client mqtt.Client, topic string, payload *[]byte) {
+	client.Publish(topic, 0, false, *payload)
+}
+
+func (r *Recorder) CurrentSteering() *events.SteeringMessage {
+	r.muSteeringMsg.Lock()
+	defer r.muSteeringMsg.Unlock()
+	steering := r.currentSteering
+	return steering
+}
+
+func (r *Recorder) Enabled() bool {
+	r.muEnabled.RLock()
+	defer r.muEnabled.RUnlock()
+	return r.enabled
+}
+
+var registerCallBacks = func(r *Recorder) {
+	err := service.RegisterCallback(r.client, r.cameraTopic, r.onFrame)
+	if err != nil {
+		log.Panicf("unable to register callback to %v:%v", r.cameraTopic, err)
+	}
+
+	err = service.RegisterCallback(r.client, r.steeringTopic, r.onSteering)
+	if err != nil {
+		log.Panicf("unable to register callback to %v:%v", r.steeringTopic, err)
+	}
+
+	err = service.RegisterCallback(r.client, r.switchRecordTopic, r.onSwitchRecord)
+	if err != nil {
+		log.Panicf("unable to register callback to %v:%v", r.switchRecordTopic, err)
+	}
+}

part/record_test.go

@@ -0,0 +1,115 @@
+package part
+
+import (
+	"github.com/cyrilix/robocar-base/testtools"
+	"github.com/cyrilix/robocar-protobuf/go/events"
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/golang/protobuf/proto"
+	"io/ioutil"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestRecorder_RecordOff(t *testing.T) {
+	oldRegister := registerCallBacks
+	oldPublish := publish
+	defer func() {
+		registerCallBacks = oldRegister
+		publish = oldPublish
+	}()
+
+	registerCallBacks = func(_ *Recorder) {}
+
+	recordTopic := "topic/record"
+	cameraTopic := "topic/camera"
+	steeringTopic := "topic/steeringMsg"
+	switchRecord := "topic/switch/record"
+
+	var muEventsPublished sync.Mutex
+	var eventsPublished *events.RecordMessage
+	publish = func(client mqtt.Client, topic string, payload *[]byte) {
+		if topic != recordTopic {
+			t.Errorf("event published on bad topic: %v, wants %v", topic, recordTopic)
+			return
+		}
+		muEventsPublished.Lock()
+		defer muEventsPublished.Unlock()
+		var msg events.RecordMessage
+		err := proto.Unmarshal(*payload, &msg)
+		if err != nil {
+			t.Errorf("unable to record plublished event: %v", err)
+		}
+		eventsPublished = &msg
+	}
+
+	recorder := NewRecorder(nil, recordTopic, cameraTopic, steeringTopic, switchRecord)
+
+	go func() {
+		if err := recorder.Start(); err == nil {
+			t.Fatalf("unable to start recorder: %v", err)
+		}
+	}()
+
+	frame1 := loadImage(t, "testdata/img.jpg", "01")
+	frame2 := loadImage(t, "testdata/img.jpg", "02")
+	steeringRight := events.SteeringMessage{Steering: 0.5, Confidence: 1.0}
+	steeringLeft := events.SteeringMessage{Steering: -0.5, Confidence: 1.0}
+
+	cases := []struct {
+		recordMsg         *events.SwitchRecordMessage
+		frameMsg          *events.FrameMessage
+		steeringMsg       *events.SteeringMessage
+		expectedRecordMsg *events.RecordMessage
+		wait              time.Duration
+	}{
+		{recordMsg: &events.SwitchRecordMessage{Enabled: false}, frameMsg: nil, steeringMsg: nil, expectedRecordMsg: nil, wait: 5 * time.Millisecond},
+		{recordMsg: &events.SwitchRecordMessage{Enabled: true}, frameMsg: nil, steeringMsg: nil, expectedRecordMsg: nil, wait: 5 * time.Millisecond},
+		{recordMsg: &events.SwitchRecordMessage{Enabled: true}, frameMsg: frame1, steeringMsg: nil, expectedRecordMsg: nil, wait: 5 * time.Millisecond},
+		{recordMsg: &events.SwitchRecordMessage{Enabled: true}, frameMsg: nil, steeringMsg: &steeringRight, expectedRecordMsg: nil, wait: 5 * time.Millisecond},
+		{recordMsg: &events.SwitchRecordMessage{Enabled: true}, frameMsg: frame1, steeringMsg: &steeringRight, expectedRecordMsg: &events.RecordMessage{Frame: frame1, Steering: &steeringRight}, wait: 5 * time.Millisecond},
+		{recordMsg: &events.SwitchRecordMessage{Enabled: true}, frameMsg: nil, steeringMsg: &steeringLeft, expectedRecordMsg: nil, wait: 5 * time.Millisecond},
+		{recordMsg: &events.SwitchRecordMessage{Enabled: true}, frameMsg: frame2, steeringMsg: &steeringLeft, expectedRecordMsg: &events.RecordMessage{Frame: frame2, Steering: &steeringLeft}, wait: 5 * time.Millisecond},
+	}
+
+	for _, c := range cases {
+		muEventsPublished.Lock()
+		eventsPublished = nil
+		muEventsPublished.Unlock()
+
+		if c.recordMsg != nil {
+			recorder.onSwitchRecord(nil, testtools.NewFakeMessageFromProtobuf(recordTopic, c.recordMsg))
+		}
+		if c.frameMsg != nil {
+			recorder.onFrame(nil, testtools.NewFakeMessageFromProtobuf(cameraTopic, c.frameMsg))
+		}
+		if c.steeringMsg != nil {
+			recorder.onSteering(nil, testtools.NewFakeMessageFromProtobuf(steeringTopic, c.steeringMsg))
+		}
+
+		time.Sleep(c.wait)
+
+		if c.expectedRecordMsg == nil && eventsPublished != nil {
+			t.Errorf("unexpected published event: %v", eventsPublished)
+		}
+		if c.expectedRecordMsg.String() != eventsPublished.String() {
+			t.Errorf("bad message published: %v, wants %v", eventsPublished, c.expectedRecordMsg)
+		}
+	}
+
+}
+
+func loadImage(t *testing.T, imgPath string, id string) *events.FrameMessage {
+	jpegContent, err := ioutil.ReadFile(imgPath)
+	if err != nil {
+		t.Fatalf("unable to load image: %v", err)
+	}
+	msg := &events.FrameMessage{
+		Id: &events.FrameRef{
+			Name: imgPath,
+			Id:   id,
+		},
+		Frame: jpegContent,
+	}
+	return msg
+}