StreamDelegate 在 swift 读取大量读数后停止接收事件

Question

我有一个通过套接字连接接收消息的iOS聊天应用程序。

当用户长时间打开应用程序并且有超过50条未读消息时，服务器通过套接字发送一条消息，告知未读消息的数量，此时应用程序显示带有进度条的警报，然后服务器发送每条消息。

因此，应用程序在 StreamDelegate 方法 stream(_ stream： Stream， handle eventCode： Stream.Event)上获取每条消息，并更新进度条，直到消息结束。

问题是，当我有大量未读消息(大约 300+)时，StreamDelegate 会停止接收带有消息的事件，并且不显示任何错误消息。

我在全局队列上调用连接方法：

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

这是我的套接字连接代码：

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

在 StreamDelegate 方法 stream(_ stream： Stream， handle eventCode： Stream.Event)中，我获取消息事件并在方法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

读取每条消息的方法：

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

有人知道如何解决它吗？

Answer 1

主要思想是在成功读取操作后强制调度流的读取：

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