如何使用 Combine 将异步加速度计更新发送到服务器

Question

我的目标是以定期有效负载的形式将异步加速度计读数发送到服务器。

加速度计数据在离线和网络请求期间同时继续，因此我需要处理网络故障以及在每个网络请求期间到达的数据。

我不优雅的方法是将每个新更新附加到一个数组中：

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

然后定期向服务器发送一组值（network 是我对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)

这种方法的主要问题是，在网络请求触发和完成之间添加的数组元素会从数组中删除，而不会发送到服务器。

我有一些关于为每个网络请求添加队列和出队 X 元素的想法（但是如果请求失败，我是否将它们添加回队列？）。

我不禁想到有更好的方法来解决这个问题，使用 Combine 将这些加速度计更新“流式传输”到某种数据结构以缓冲它们，然后将它们发送到服务器。

postAccelerometerData() 函数只是对 JSON 结构进行编码并发出网络请求。那里没什么特别的。

Answer 1

Combine 有一种方法可以在一定时间内收集值并发出一个数组。因此，您可以围绕该方法编排您的解决方案，使用PassthroughSubject 发送每个值，并使用.collect 运算符和byTime 策略将值收集到一个数组中。

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) 

以上是一个简化的示例 - 它不处理加速度计错误或网络错误 - 因为它似乎超出了您在这个问题中的要求。但是，如果您需要处理网络错误并重试 - 那么您可以将 postAccelerometerData 包装在 Future 中并将其集成到组合管道中。

Answer 2

已编辑

另一种选择是使用throttle(for:scheduler:latest:) 和@Published dataArray：

假设postAccelerometerData(payload:) 仅显示 data，让我们考虑以下 - 自愿 - 简单视图：

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())
    }
}

为避免过于频繁地更新视图，最好将运动管理器包装在一个单独的类中：

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
        }
    }
}

现在ViewModel 只听$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"
    }
}

这里的主要思想是通过这样做：

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

您每次收到更新时都会写入data，因此如果是@Published，您可以简单地使用它的发布者和throttle，按照您选择的时间间隔将更新发布到服务器。当然，如果您的网络层需要处理错误并重试，它仍然是可能的，但最好将其单独处理