Swift Combine: Publisher for recording BLE peripheral data stream

Question

Premis. I have a BLE (Bluetooth low energy) peripheral that will stream some data that it has recorded. I am using the BlueJay (https://github.com/steamclock/bluejay) package to interface with the BLE peripheral.

the peripheral will start streaming after I write to its "startDownload" characteristic, I will also need to subscribe to its download service before calling "startDownload" characteristic.

Goal: I want to do this using combine as it will make my life easier as the app is made using SwiftUI. I have started to use combine recently and I am not as proficient with it as I would like to be (still a combine noob, leaving the world of old UIKit is hard :) ).

So BluJay will update my code using a callback closure and I want to wrap it in a publisher that will send an update every time I get a new block of data in. Bonus would be to add a progress report somewhere in the chain of operators.

this is how I would like it to be used

BLEmanager.startDownload()
   .someOperator{ dataBlock in
       //validate that block is valid 
   }
   .someOperator{ progress in // optional and not currently needed
       self.progress = progress
   }
   .sink{ completion in
       switch completion{
       case .finished:
           //update ui as download is completed
       case .failure(let error):
           // deal with error
   } receiveValue:{
        //add the data to already received data
   }.store(in: &storage) 

(Edit: added the receiveValue block)

ok so a BlueJay IPA looks like this :

let characteristic = CharacteristicIdentifier(uuid: BLEModel.FileDownload.Characteristics.dataUUID,
                                                      service: ServiceIdentifier(uuid: BLEModel.FileDownloadInfo.serviceUUID))
    
        startListen(characteristic) {(result: ReadResult<Data>) in
            switch result {
            case .success(_):
                //we got new data to process and concatenated to previously received data. 
            case .failure(let error):
                // fire error
            }
        }

and this code would sit inside the startDownload() method. every time a new block of data is received we report it to the downstream publisher and then it gets validated, then again passed down the chain, optionally the progress is reported and then the new block is added to the previous data collected.

I saw this Swift Combine Future with multiple values? but it only sends progress and internally stacks the frames/data. This is kind of close to what I need but I am not sure how to change it to make it do what I need it to do.

Initially, I was hoping to be able to just use Future to report on every update but it looks like this is only good for 1 shot of data.

func startDownload() -> AnyPublisher<Data, Error> {
    Future{ callback in
        // BlueJay API calls 
        // but calling callback(.success(newData)) would as I understand it 
        // stop after the first time I report a success.. 
    }
    ereaseToAnyPublisher()
}

So I need to bridge from making a simple report using a Future publisher to making some streaming publisher.

Edit (09/09/2021):

So following New Dev's comment I coded this function.

func startListenFileTransfer(rideSize size: UInt32, blockSize block: UInt32 = 200 ) -> AnyPublisher<Data, Error>{
        let publisher = PassthroughSubject<Data, Error>()
        let characteristic = CharacteristicIdentifier(uuid: BikeBLEModel.FileDownloadInfo.Characteristics.dataUUID,
                                                      service: ServiceIdentifier(uuid: BikeBLEModel.FileDownloadInfo.serviceUUID))
        var fetchPageStartIndex: UInt32 = 0
        setFileDownloadStart()
        setReadSegment()
            startListen(characteristic) {[unowned self](result: ReadResult<Data>) in
                debugPrint("DEBUG listen result in page \(fetchPageStartIndex + 1)")
                switch result {
                case .success(let data):
                    publisher.send(data)
                    fetchPageStartIndex += 1
                    if fetchPageStartIndex >= size {
                        debugPrint("DEBUG download done page recieved \(fetchPageStartIndex)")
                        publisher.send(completion: Subscribers.Completion.finished)
                        stopFileDownload()
                    }
                    else if fetchPageStartIndex % block == 0 {
                        fetchPageStartIndex += 1
                        debugPrint("DEBUG: - set meme segment at \(fetchPageStartIndex)")
                        setReadSegment(fetchPageStartIndex)
                    }
                case .failure(let error):
                    debugPrint("DEBUG Error: \(error)")
                    publisher.send(completion: Subscribers.Completion.failure(error))
                    stopFileDownload()
                }
            }
                        
        return publisher.eraseToAnyPublisher()
    }