StreamDelegate s'arrête pour recevoir des événements après un certain nombre de lectures dans swift

Question

J'ai une application de chat iOS qui reçoit des messages via une connexion socket.

Lorsque l'utilisateur ouvre l'application après une longue période et qu'il y a plus de 50 messages non lus, le serveur envoie un message indiquant le nombre de messages non lus, à ce stade, l'application affiche une alerte avec une barre de progression. le serveur envoie chaque message. Ainsi, l'application reçoit chaque message sur la méthode StreamDelegate flux (_ stream: Stream, gérer eventCode: Stream.Event) et met à jour la barre de progression jusqu'à la fin des messages. Le problème est que, quand j'ai une grande quantité de messages non lus (environ 300+) à un certain point, le StreamDelegate s'arrête pour recevoir les événements avec les messages, et aucun message d'erreur est affiché.

J'appelle la méthode de connexion sur une file d'attente globale:

DispatchQueue.global().async { 
    self.connect(host, port: port) 
} 

Ceci est mon connexion socket code:

fileprivate func connect(_ host: String, port: Int) { 

     postStatus(.connecting) 

     self.host = NSString(string: host) 
     self.port = UInt32(port) 

     self.log("connect to \(host):\(port)") 

     var readStream : Unmanaged<CFReadStream>? 
     var writeStream : Unmanaged<CFWriteStream>? 

     CFStreamCreatePairWithSocketToHost(nil, self.host, self.port, &readStream, &writeStream) 

     self.inOk = false 
     self.outOk = false 
     self.inputStream = readStream!.takeRetainedValue() 
     self.outputStream = writeStream!.takeRetainedValue() 

     self.inputStream.delegate = self 
     self.outputStream.delegate = self 


     let mainThread = Thread.isMainThread; 

     let loop = mainThread ? RunLoop.main : RunLoop.current 

     self.inputStream.schedule(in: loop, forMode: RunLoopMode.defaultRunLoopMode) 
     self.outputStream.schedule(in: loop, forMode: RunLoopMode.defaultRunLoopMode) 

     self.inputStream.open() 
     self.outputStream.open() 

     self.timer = Timer.scheduledTimer(timeInterval: 5, target: self, selector: #selector(connectionTimeout), userInfo: nil, repeats: false) 

     if(!mainThread) { 
      loop.run() 
     } 

    } 

Dans le courant de la méthode StreamDelegate (_ flux: flux, poignée eventCode: Stream.Event) Je reçois l'événement de message et le traite sur la méthode read (String)

case Stream.Event.hasBytesAvailable: 

     if let timer = timer { 
      timer.invalidate() 
      self.timer = nil 
     } 

     let json = ChatLibSwift.readMessage(self.inputStream) 

     do { 
      if StringUtils.isNotEmpty(json) { 
       try self.read(json) 
      } 
     } catch let ex as NSError { 
      LogUtils.log("ERROR: \(ex.description)") 
     } 

     break 
    case Stream.Event.hasSpaceAvailable: 
     break 

La méthode qui a lu chacun message:

static func readMessage(_ inputStream: InputStream) -> String { 

    do { 
     var lenBytes = [UInt8](repeating: 0, count: 4) 


     inputStream.read(&lenBytes, maxLength: 4) 

     // header 

     let i32: Int = Int(UInt32.init(lenBytes[3]) | UInt32.init(lenBytes[2]) << 8 | UInt32.init(lenBytes[1]) << 16 | UInt32.init(lenBytes[0]) << 24) 

     var msg = [UInt8](repeating: 0, count: (MemoryLayout<UInt8>.size * Int(i32))) 

     let bytesRead = inputStream.read(&msg, maxLength: Int(i32)) 

     if bytesRead == -1 { 
      print("<< ChatLib ERROR -1") 
      return "" 
     } 

     let s = NSString(bytes: msg, length: bytesRead, encoding: String.Encoding.utf8.rawValue) as String? 

     if let s = s { 
      if bytesRead == Int(i32) { 
       return s 
      } 
      else { 
       print("Error: readMessage \(s)") 
      } 
      return s 
     } 
     return "" 
    } catch { 

     return "" 
    } 
} 

Toute personne a idée de comment résoudre?

Answer 1

L'idée principale est de forcer le calendrier de la lecture du flux après une opération de lecture réussie:

let _preallocatedBufferSize = 64 * 1024 
var _preallocatedBuffer = [UInt8](repeating: 0, count: MemoryLayout<UInt8>.size * Int(_preallocatedBufferSize)) 

var message : .... 

func readMessage(_ inputStream: InputStream) { 

    if !inputStream.hasBytesAvailable || message.isCompleted { 
     return 
    } 

    var theBuffer : UnsafeMutablePointer<UInt8>? 
    var theLength : Int = 0 

    // try to get buffer from the stream otherwise use the preallocated buffer 
    if !inputStream.getBuffer(&theBuffer, length:&theLength) || nil == theBuffer 
    { 
     memset(&_preallocatedBuffer, 0, _preallocatedBufferSize) 

     let theReadCount = inputStream.read(&_preallocatedBuffer, maxLength:_preallocatedBufferSize) 
     if theReadCount > 0 { 
      theBuffer = _preallocatedBuffer; 
      theLength = theReadCount; 
     } else { 
      theBuffer = nil; 
      theLength = 0; 
     } 
    } 

    if nil != theBuffer && theLength > 0 { 
     _message.appendData(theBuffer, length:theLength) 

     self.perform(#selector(readMessage), with:inputStream, afterDelay:0.0, inModes:[RunLoopMode.defaultRunLoopMode]) 
    } 
}