swift - Swift 字节中的 UDP 消息已读但无消息

Question

我正在使用 Swift 编写这个 UDP 客户端示例。我有一个桥接 header 来合并适当的 C header 文件。在示例程序中，我可以看到正在发送和接收的字节数是正确的。但是，当我尝试将消息打印到控制台时，我什么也得不到。应该有一条消息出现红色箭头:

我愿意打赌这是一件小事，但从昨天晚上开始我就一直在研究这个，我已经没有想法了。有什么建议吗？

UDP客户端:

import Foundation

enum UDPClientError: Int, LocalizedError {

case noSocket = -999
case bindSocket
case badAddress
case alreadyInProgress
case setupForNonBlocking
case threadLock

var localizedDescription: String {

    switch self {

    case .alreadyInProgress:
        return "operation in progress"
    case .badAddress:
        return "Address string given is not valid"
    case .bindSocket:
        return "Could not bind socket"
    case .noSocket:
        return "Could not obtain socket"
    case .setupForNonBlocking:
        return "Could not setup socket for non-blocking operation"
    case .threadLock:
        return "Could not obtain thread lock"
    }

}
}

class UDPClient {

private var mySocket: Int32 = 0

private var myAddress = sockaddr_in()

private var otherAddress = sockaddr_in()

let name: String

private var receiveQueue = [String]()

private var sendQueue = [String]()

private var okToRun = false

private var threadLock = pthread_mutex_t()

init(name: String, port: UInt16, otherIPAddress: String, otherPort: UInt16) throws {

    self.name = name

    mySocket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)

    if mySocket == -1 {

        throw UDPClientError.noSocket
    }

    if fcntl(mySocket, F_SETFL, O_NONBLOCK) == -1 {

        throw UDPClientError.setupForNonBlocking
    }

    myAddress.sin_family = sa_family_t(AF_INET)
    myAddress.sin_port = in_port_t(port)
    myAddress.sin_addr.s_addr = in_addr_t(INADDR_ANY)

    let retCode = withUnsafeMutablePointer(to: &myAddress) {

        $0.withMemoryRebound(to: sockaddr.self, capacity: 1) {

            bind(mySocket, UnsafeMutablePointer<sockaddr>($0), socklen_t(MemoryLayout<sockaddr_in>.size))
        }
    }

    if retCode == -1 {

        throw UDPClientError.bindSocket
    }

    otherAddress.sin_family = sa_family_t(AF_INET)
    otherAddress.sin_port = in_port_t(otherPort)

    var buffer: [Int8] = Array(otherIPAddress.utf8CString)

    if inet_aton(&buffer, &otherAddress.sin_addr) == 0 {

        throw UDPClientError.badAddress
    }

    if pthread_mutex_init(&threadLock, nil) != 0 {

        throw UDPClientError.threadLock
    }

    print("done")
}


func beginOperation() {

    okToRun = true
    _ = Thread.detachNewThreadSelector(#selector(process), toTarget: self, with: nil)
    //processThread.start()

}

func endOperation() {

    okToRun = false
}

func send(message: String) {

    pthread_mutex_lock(&threadLock)

    sendQueue.append(message)

    pthread_mutex_unlock(&threadLock)

}

func getMessage() -> String? {

    pthread_mutex_lock(&threadLock)

    let flag = receiveQueue.isEmpty

    pthread_mutex_unlock(&threadLock)

    if flag {
        print("no message")
        return nil
    }

    pthread_mutex_lock(&threadLock)

    let message = receiveQueue.remove(at: 0)

    pthread_mutex_unlock(&threadLock)

    return message
}

@objc private func process() {

    //let bufferLimit = 1024
    var buffer = [UInt8](repeating: 0, count: 1024)
    buffer.removeAll(keepingCapacity: true)

    var slen = socklen_t(MemoryLayout<sockaddr_in>.size)

    print("Process running for " + name)

    var bytesRead = 0
    var bytesSent = 0

    while okToRun {

        if sendQueue.isEmpty == false {

            buffer.removeAll(keepingCapacity: true)

            pthread_mutex_lock(&threadLock)

            buffer.append(contentsOf: sendQueue.remove(at: 0).utf8)

            pthread_mutex_unlock(&threadLock)

            bytesSent = withUnsafeMutablePointer(to: &otherAddress) {

                $0.withMemoryRebound(to: sockaddr.self, capacity: 1) {

                    sendto(mySocket, buffer, buffer.count, 0, UnsafePointer<sockaddr>($0), socklen_t(MemoryLayout<sockaddr_in>.size))
                }
            }

            if bytesSent != -1 {

                print("First buffer character: \(buffer[0])")
                print("\(name): sendto() bytes sent: \(bytesSent)")
            }

            buffer.removeAll(keepingCapacity: true)
        }

        bytesRead = withUnsafeMutablePointer(to: &otherAddress) {

            $0.withMemoryRebound(to: sockaddr.self, capacity: 1) {

                recvfrom(mySocket, UnsafeMutableRawPointer(mutating: buffer), 1024, 0, UnsafeMutablePointer<sockaddr>($0), &slen)
            }
        }

        if bytesRead != -1 {

            print("\(name): Bytes read = \(bytesRead) bytes: \(buffer)")

            pthread_mutex_lock(&threadLock)

            receiveQueue.append(String(bytes: buffer, encoding: .utf8)!)

            pthread_mutex_unlock(&threadLock)

            slen = socklen_t(MemoryLayout<sockaddr_in>.size)
        }

        bytesRead = 0
        bytesSent = 0

    } // end processing loop

} // end process

}

UDP 桥接 header :

#ifndef UDP_Bridging_Header_h
#define UDP_Bridging_Header_h

#include <arpa/inet.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <pthread.h>

#endif /* UDP_Bridging_Header_h */

试用客户端的主要文件:

import Foundation

var client_1: UDPClient?
var client_2: UDPClient?


do {

    try client_1 = UDPClient(name: "Client A", port: 9990, otherIPAddress: "192.168.2.4", otherPort: 9992)
}
catch {

    print(error.localizedDescription)
}

do {

    try client_2 = UDPClient(name: "Client B", port: 9992, otherIPAddress: "192.168.2.4", otherPort: 9990)
}
catch {

    print(error.localizedDescription)
}

if client_1 != nil && client_2 != nil {

    client_1!.send(message: "Try this out")

    client_1!.beginOperation()
    client_2!.beginOperation()

    Thread.sleep(forTimeInterval: 5.0)

    if let msg = client_2!.getMessage() {

        print(msg)
    }
}

Answer 1

问题是您在调用 recvfrom 之前清空了数组:

buffer.removeAll(keepingCapacity: true)
// ...
recvfrom(mySocket, UnsafeMutableRawPointer(mutating: buffer), 1024, 0, UnsafeMutablePointer<sockaddr>($0), &slen)

recvfrom 读入一个内存位置，但对Swift Array 类型，不会附加数组元素或更新其计数。

你必须传递一个指向非空元素存储的指针数组代替。例如，要从套接字中读取最多 1024 个字节:

buffer = [UInt8](repeating: 0, count: 1024)
bytesRead = withUnsafeMutablePointer(to: &otherAddress) {
    $0.withMemoryRebound(to: sockaddr.self, capacity: 1) {
        recvfrom(mySocket, &buffer, buffer.count, 0, $0, &slen)
    }
}
if bytesRead != -1 {
    buffer.removeSubrange(bytesRead ..< buffer.count)
    // process buffer ...

}

注意 UnsafeMutableRawPointer(mutating: buffer) 可以简化到调用中的 &buffer。