I have a table called PendingFile, which holds the parts of a file being downloaded through Socket. Every, pendingFile has a status = FILE_PART_PENDING || FILE_PART_RECEIVED.

PendingFile

@Entity(tableName = "pendingFile", primaryKeys = ["m_id"])
data class PendingFile(
    @ColumnInfo(name = "m_id")
    val mId: Long,
    @ColumnInfo(name = "file_hash")
    val hash: String,
    @ColumnInfo(name = "message_id")
    val messageId: String,
    @ColumnInfo(name = "segment")
    val segment: Int,
    @ColumnInfo(name = "total")
    val total: Int,
    @ColumnInfo(name = "status")
    val status: Int = FILE_PART_PENDING,
    @ColumnInfo(name = "last_event_date")
    val date: String
)

I also have a query in the Dabatase to get the completeCount parts of the File (hash)

// Calculates how many parts have been downloaded
@Query("""Select count(*) from pendingFile where file_hash = :fileHash and status = :status""")
suspend fun getFilePartCountByStatus(fileHash: String, status: Int): Int

The function which called when a part received from the Socket

suspend fun onFilePartReceived(args...) {
   val filePart = filesRepository.getFilePartBySegment(receiver.filePart.fileHash, receiver.filePart.segment) ?: return
      if (filePart.status == FILE_PART_RECEIVED) return

      // Take the part and create a file
      println("## File part received ${receiver.filePart.segment}")
      filesRepository.createAndWriteFilePart(server.serverId, receiver.filePart.fileHash, receiver.filePart.segment, receiver.filePart.data)

      // Mark the part as Received in the Database
      filesRepository.updateFilePartStatus(server.serverId, receiver.filePart.fileHash, receiver.filePart.segment, FILE_PART_RECEIVED)


      // Check if it is the last one Received, in order to proceed
      val completeCount = filesRepository.getFilePartCountByStatus(receiver.filePart.fileHash, FILE_PART_RECEIVED)
      if (completeCount == receiver.filePart.total) {
          println("## Complete -> $completeCount parts received")
      } 
}

There is also a MessageHandler singleton class which has a CoroutineScope and if the incoming message is a FilePart then it called the onFilePartReceived method.

class MessageHandlerImpl : MessageHandler , KoinComponent {

   private val scope = CoroutineScope(Dispatchers.Main + SupervisorJob())

   fun messageResolver(message: String, socketCallback: (message: String) -> Unit) {
        scope.launch(Dispatchers.IO) {
            ....
            if (message.isFilePart())
               filesHandler.onFilePartReceived(args...)
        }
}

The problem that i am facing here is this. When i ask from the Socket server to send me parts (1..500), i want the onFilePartReceived to get called every time a new part comes in parallel. So all the parts can be handled at the same time. And the second issue is that i see in my Logcat the println("## Complete -> ....") more than one times. That means that there is a race condition between the many instances of the onFilePartReceived function. How can i solve this?