Add raw command and clear device queue endpoints (#864)

platform/queue/inmem/inmem.go

@@ -33,6 +33,7 @@ func New(pubsub pubsub.PublishSubscriber, logger log.Logger) *QueueInMem {
 		queue:  make(map[string]*list.List),
 	}
 	q.startPolling(pubsub)
+	q.startRawPolling(pubsub)
 	return q
 }
 
@@ -169,3 +170,44 @@ func (q *QueueInMem) startPolling(pubsub pubsub.PublishSubscriber) error {
 	}()
 	return nil
 }
+
+func (q *QueueInMem) startRawPolling(pubsub pubsub.PublishSubscriber) error {
+	events, err := pubsub.Subscribe(context.TODO(), "command-queue", command.RawCommandTopic)
+	if err != nil {
+		return err
+	}
+	go func() {
+		for {
+			select {
+			case event := <-events:
+				var cmdEvent command.RawEvent
+				if err := command.UnmarshalRawEvent(event.Message, &cmdEvent); err != nil {
+					level.Info(q.logger).Log(
+						"msg", "unmarshal command event from pubsub",
+						"err", err,
+					)
+					continue
+				}
+				q.enqueue(
+					q.getList(cmdEvent.DeviceUDID),
+					cmdEvent.CommandUUID,
+					cmdEvent.Payload,
+				)
+				level.Info(q.logger).Log(
+					"msg", "queued raw command for device",
+					"device_udid", cmdEvent.DeviceUDID,
+					"command_uuid", cmdEvent.CommandUUID,
+				)
+
+				err = boltqueue.PublishCommandQueued(pubsub, cmdEvent.DeviceUDID, cmdEvent.CommandUUID)
+				if err != nil {
+					level.Info(q.logger).Log(
+						"msg", "publish command to queued topic",
+						"err", err,
+					)
+				}
+			}
+		}
+	}()
+	return nil
+}

platform/queue/queue.go

@@ -212,6 +212,10 @@ func NewQueue(db *bolt.DB, pubsub pubsub.PublishSubscriber, opts ...Option) (*St
 		return nil, err
 	}
 
+	if err := datastore.pollRawCommands(pubsub); err != nil {
+		return nil, err
+	}
+
 	return datastore, nil
 }
 
@@ -318,6 +322,58 @@ func (db *Store) pollCommands(pubsub pubsub.PublishSubscriber) error {
 	return nil
 }
 
+func (db *Store) pollRawCommands(pubsub pubsub.PublishSubscriber) error {
+	commandEvents, err := pubsub.Subscribe(context.TODO(), "command-queue", command.RawCommandTopic)
+	if err != nil {
+		return errors.Wrapf(err,
+			"subscribing push to %s topic", command.RawCommandTopic)
+	}
+	go func() {
+		for {
+			select {
+			case event := <-commandEvents:
+				var ev command.RawEvent
+				if err := command.UnmarshalRawEvent(event.Message, &ev); err != nil {
+					level.Info(db.logger).Log("msg", "unmarshal raw command event in queue", "err", err)
+					continue
+				}
+
+				cmd := new(DeviceCommand)
+				cmd.DeviceUDID = ev.DeviceUDID
+				byUDID, err := db.DeviceCommand(ev.DeviceUDID)
+				if err == nil && byUDID != nil {
+					cmd = byUDID
+				}
+				newCmd := Command{
+					UUID:    ev.CommandUUID,
+					Payload: ev.Payload,
+				}
+				cmd.Commands = append(cmd.Commands, newCmd)
+				if err := db.Save(cmd); err != nil {
+					level.Info(db.logger).Log("msg", "save command in db", "err", err)
+					continue
+				}
+				level.Info(db.logger).Log(
+					"msg", "queued raw event for device",
+					"device_udid", ev.DeviceUDID,
+					"command_uuid", ev.CommandUUID,
+				)
+
+				err = PublishCommandQueued(pubsub, ev.DeviceUDID, ev.CommandUUID)
+				if err != nil {
+					level.Info(db.logger).Log(
+						"msg", "publish command to queued topic",
+						"err", err,
+					)
+					continue
+				}
+			}
+		}
+	}()
+
+	return nil
+}
+
 func isNotFound(err error) bool {
 	if _, ok := err.(*notFound); ok {
 		return true