How to use Combine to send async accelerometer updates to server

user_577237 0
Viewed 170

My goal is to send async accelerometer readings to a server in periodic payloads.

Accelerometer data continues while offline and concurrently during the network requests, so I'll need to handle network failures as well as data that arrives during the duration of each network request.

My inelegant approach is to append each new update to an array:

motionManager.startAccelerometerUpdates(to: .main) { data, error in
    dataArray.append(data)
}

And then periodically send a group of values to the server (network is my wrapper around NWPathMonitor()):

let timer = Timer(fire: Date(), interval: 5, // Every 5 seconds
              repeats: true, block: { timer in
                if network.connected {
                    postAccelerometerData(payload: dataArray) { success
                        if success {
                            dataArray.removeAll()
                        }
                    }
                }
            })


RunLoop.current.add(timer, forMode: RunLoop.Mode.default)

The major issue with this approach is that the elements of the array added between when the network request fires and when it completes would be removed from the array without ever being sent to the server.

I've had some ideas about adding a queue and dequeuing X elements on for each network request (but then do I add them back to the queue if the request fails?).

I can't help but think there is a better way to approach this using Combine to "stream" these accelerometer updates to some sort of data structure to buffer them, and then send those on to a server.

The postAccelerometerData() function just encodes a JSON structure and makes the network request. Nothing particularly special there.

combine sensors swift buffer networking
2 Answers

Combine has a way to collect values for a certain amount of time and emit an array. So, you could orchestrate your solution around that approach, by using a PassthroughSubject to send each value, and the .collect operator with byTime strategy to collect the values into an array.

let accelerometerData = PassthroughSubject<CMAccelerometerData, Never>()

motionManager.startAccelerometerUpdates(to: .main) { data, error in
   guard let data = data else { return } // for demo purposes, ignoring errors
   accelerometerData.send(data)
}

// set up a pipeline that periodically sends data to the server
accelerometerData
   .collect(.byTime(DispatchQueue.main, .seconds(5))) // collect for 5 sec
   .sink { dataArray in
       // send to server
       postAccelerometerData(payload: dataArray) { success in
           print("success:", success)
       }
   }
   .store(in: &cancellables)

The above is a simplified example - it doesn't handle accelerometer errors or network errors - because it seems to be beyond what you're asking in this question. But if you need to handle network errors and, say, retry - then you could wrap postAccelerometerData in a Future and integrate it into the Combine pipeline.

user_9681550

Edited

Another option would be to use throttle(for:scheduler:latest:) and a @Published dataArray:

Let's say that postAccelerometerData(payload:) only displays the data and let's consider the following - voluntarily - simple view :

struct ContentView: View {
    @ObservedObject var viewModel = ViewModel()
    var body: some View {
        Text(viewModel.description)
            .padding()
    }
}

struct ContentView_Previews: PreviewProvider {
    static var previews: some View {
        ContentView(viewModel: ViewModel())
    }
}

To avoid updating the view too frequently it's better wrap the motion manager in a separate class :

class MotionManagerObserver {
    var motionManager: CMMotionManager
    var cancellables: Set<AnyCancellable> = []
    @Published var data: CMAccelerometerData? = nil
    init() {
        motionManager = CMMotionManager()
        // "start" the accelerometer
        motionManager.startAccelerometerUpdates(to: .main) { data, _  in
            self.data = data
        }
    }
}

now ViewModel only listens to the $dataArray :

class ViewModel: ObservableObject, CustomStringConvertible {
    var cancellables: Set<AnyCancellable> = []
    let observer: MotionManagerObserver
    @Published var description: String
    
    init() {
        description = ""
        observer = MotionManagerObserver()
        // the magic happens here with `throttle(for:scheduler:latest:)`
        // `postAccelerometerData(payload:)` will be called every 5s
        // and the `Text` view will be updated
        observer.$data
            .throttle(for: 5, scheduler: RunLoop.main, latest: true)
            .sink(receiveValue: postAccelerometerData(payload:))
            .store(in: &cancellables)
    }
  
    func postAccelerometerData(payload: CMAccelerometerData?) {
        description = payload?.description ?? "N/A"
    }
}

The main idea here is that by doing:

motionManager.startAccelerometerUpdates(to: .main) { data, _  in
    self.data = data
}

you write into data each time you receive an update so if it's @Published you can simply use it's publisher and throttle to post updates to the server at the interval you choose. And of course if you network layer needs to handle errors and retry it is still possible but it is probably a good idea to keep that a separate matter

user_14256970
Related