starting resource subscribers implementation

basestation.go

@@ -127,9 +127,7 @@ func (b *Basestation) makeEventStreamRequest(ctx context.Context, payload EventS
 	transId := genTransId()
 	payload.TransId = transId
 
-	responseChan := make(chan *EventStreamResponse)
+	responseChan := b.eventStream.subscribeTransaction(transId)
-	b.eventStream.subscribe(transId, responseChan)
-	defer b.eventStream.unsubscribe(transId)
 
 	// Send the payload to the event stream.
 	if err := b.NotifyEventStream(payload); err != nil {
@@ -166,16 +164,12 @@ func (b *Basestation) Subscribe(ctx context.Context) error {
 	if err != nil {
 		return fmt.Errorf("setting up event stream: %v", err)
 	}
-forLoop:
+	select {
-	for {
+	case <-ctx.Done():
-		select {
+		return fmt.Errorf("failed to subscribe to the event stream: requesting shutdown")
-		case <-ctx.Done():
+	case connected := <-connectedChan:
-			return fmt.Errorf("failed to subscribe to the event stream: requesting shutdown")
+		if !connected {
-		case connected := <-connectedChan:
+			return fmt.Errorf("failed to subscribe to the event stream")
-			if !connected {
-				return fmt.Errorf("failed to subscribe to the event stream")
-			}
-			break forLoop
 		}
 	}
 

camera.go

@@ -154,26 +154,23 @@ func (c *Camera) SetBrightness(ctx context.Context, brightness int) (response *E
 	return b.makeEventStreamRequest(ctx, payload)
 }
 
-func (c *Camera) EnableMotionAlerts(ctx context.Context, sensitivity int, zones []string) (response *EventStreamResponse, err error) {
+func (c *Camera) EnableMotionAlerts(ctx context.Context, sensitivity int, zones []string) error {
-	payload := EventStreamPayload{
-		Action:          "set",
-		Resource:        fmt.Sprintf("cameras/%s", c.DeviceId),
-		PublishResponse: true,
-		Properties: MotionDetectionProperties{
-			BaseDetectionProperties: BaseDetectionProperties{
-				Armed:       true,
-				Sensitivity: sensitivity,
-				Zones:       zones,
-			},
-		},
-		From: fmt.Sprintf("%s_%s", c.UserId, TransIdPrefix),
-		To:   c.ParentId,
-	}
 	b := c.arlo.Basestations.Find(c.ParentId)
 	if b == nil {
-		return nil, fmt.Errorf("basestation (%s) not found for camera (%s)", c.ParentId, c.DeviceId)
+		return fmt.Errorf("basestation (%s) not found for camera (%s)", c.ParentId, c.DeviceId)
 	}
-	return b.makeEventStreamRequest(ctx, payload)
+	err := b.makeRequest(ctx, "set", fmt.Sprintf("cameras/%s", c.DeviceId), true, MotionDetectionProperties{
+		BaseDetectionProperties: BaseDetectionProperties{
+			Armed:       true,
+			Sensitivity: sensitivity,
+			Zones:       zones,
+		},
+	}, nil)
+
+	if err != nil {
+		return err
+	}
+	return nil
 }
 
 func (c *Camera) DisableMotionAlerts(ctx context.Context, sensitivity int, zones []string) (response *EventStreamResponse, err error) {

cmd/main.go

@@ -18,7 +18,7 @@ func main() {
 
 	for _, device := range a.Cameras {
 		log.Infof("%s", device.DeviceName)
-		err := device.On(ctx)
+		err := device.EnableMotionAlerts(ctx, 70, nil)
 		if err != nil {
 			log.Errorf("setting camera %s on: %v", device.DeviceName, err)
 		}

events_stream.go

@@ -22,6 +22,9 @@ type eventStream struct {
 
 	transactionSubscribers      map[string]chan *EventStreamResponse
 	transactionSubscribersMutex sync.RWMutex
+
+	resourceSubscribers      map[string][]chan *EventStreamResponse
+	resourceSubscribersMutex sync.RWMutex
 }
 
 func newEventStream(url string, client *http.Client) *eventStream {
@@ -29,6 +32,8 @@ func newEventStream(url string, client *http.Client) *eventStream {
 		Events:                      make(chan *sse.Event),
 		transactionSubscribers:      make(map[string]chan *EventStreamResponse),
 		transactionSubscribersMutex: sync.RWMutex{},
+		resourceSubscribers:         make(map[string][]chan *EventStreamResponse),
+		resourceSubscribersMutex:    sync.RWMutex{},
 		disconnectedChan:            make(chan interface{}),
 		once:                        new(sync.Once),
 	}
@@ -58,6 +63,11 @@ func (e *eventStream) listen(ctx context.Context) (chan bool, error) {
 	connectedChan := make(chan bool)
 
 	go func() {
+		defer func() {
+			e.clearResourceSubscriptions()
+			e.clearTransactionSubscriptions()
+		}()
+
 		for {
 			select {
 			case <-ctx.Done():
@@ -106,6 +116,16 @@ func (e *eventStream) listen(ctx context.Context) (chan bool, error) {
 				e.transactionSubscribersMutex.RUnlock()
 				if ok {
 					subscriber <- &notifyResponse
+					e.removeTransactionSubscription(notifyResponse.TransId)
+				}
+
+				e.resourceSubscribersMutex.RLock()
+				subscribers, ok := e.resourceSubscribers[notifyResponse.Resource]
+				e.resourceSubscribersMutex.RUnlock()
+				if ok {
+					for _, s := range subscribers {
+						s <- &notifyResponse
+					}
 				}
 
 			case <-e.disconnectedChan:
@@ -118,17 +138,48 @@ func (e *eventStream) listen(ctx context.Context) (chan bool, error) {
 	return connectedChan, nil
 }
 
-func (e *eventStream) unsubscribe(transId string) {
+func (e *eventStream) subscribeTransaction(transId string) chan *EventStreamResponse {
+	msgchan := make(chan *EventStreamResponse)
 	e.transactionSubscribersMutex.Lock()
-	if c, ok := e.transactionSubscribers[transId]; ok {
+	e.transactionSubscribers[transId] = msgchan
-		close(c)
-		delete(e.transactionSubscribers, transId)
-	}
 	e.transactionSubscribersMutex.Unlock()
+	return msgchan
 }
 
-func (e *eventStream) subscribe(transId string, subscriber chan *EventStreamResponse) {
+func (e *eventStream) subscribeResource(resource string) chan *EventStreamResponse {
-	e.transactionSubscribersMutex.Lock()
+	msgchan := make(chan *EventStreamResponse)
-	e.transactionSubscribers[transId] = subscriber
+	e.resourceSubscribersMutex.Lock()
-	e.transactionSubscribersMutex.Unlock()
+	e.resourceSubscribers[resource] = append(e.resourceSubscribers[resource], msgchan)
+	e.resourceSubscribersMutex.Unlock()
+	return msgchan
+}
+
+func (e *eventStream) removeTransactionSubscription(transId string) {
+	e.transactionSubscribersMutex.Lock()
+	defer e.transactionSubscribersMutex.Unlock()
+	subscriber, ok := e.transactionSubscribers[transId]
+	if ok {
+		close(subscriber)
+		delete(e.transactionSubscribers, transId)
+	}
+}
+
+func (e *eventStream) clearResourceSubscriptions() {
+	e.resourceSubscribersMutex.Lock()
+	defer e.resourceSubscribersMutex.Unlock()
+	for _, subscribers := range e.resourceSubscribers {
+		for _, subscriber := range subscribers {
+			close(subscriber)
+		}
+	}
+	e.resourceSubscribers = make(map[string][]chan *EventStreamResponse)
+}
+
+func (e *eventStream) clearTransactionSubscriptions() {
+	e.transactionSubscribersMutex.Lock()
+	defer e.transactionSubscribersMutex.Unlock()
+	for _, subscriber := range e.transactionSubscribers {
+		close(subscriber)
+	}
+	e.transactionSubscribers = make(map[string]chan *EventStreamResponse)
 }