Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Support swiftnio #9

Merged
merged 2 commits into from
May 5, 2024
Merged

Support swiftnio #9

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e7ae515
extract network part to protcol
huiping192 Sep 7, 2023
76f1b4e
add swiftnio implement
huiping192 Dec 17, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 32 additions & 0 deletions Package.resolved
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
{
"pins" : [
{
"identity" : "swift-atomics",
"kind" : "remoteSourceControl",
"location" : "https://github.com/apple/swift-atomics.git",
"state" : {
"revision" : "cd142fd2f64be2100422d658e7411e39489da985",
"version" : "1.2.0"
}
},
{
"identity" : "swift-collections",
"kind" : "remoteSourceControl",
"location" : "https://github.com/apple/swift-collections.git",
"state" : {
"revision" : "a902f1823a7ff3c9ab2fba0f992396b948eda307",
"version" : "1.0.5"
}
},
{
"identity" : "swift-nio",
"kind" : "remoteSourceControl",
"location" : "https://github.com/apple/swift-nio",
"state" : {
"revision" : "702cd7c56d5d44eeba73fdf83918339b26dc855c",
"version" : "2.62.0"
}
}
],
"version" : 2
}
11 changes: 5 additions & 6 deletions Package.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,23 +5,22 @@ import PackageDescription

let package = Package(
name: "HPRTMP",
platforms: [.iOS(.v14),.macOS(.v11)],
platforms: [.iOS(.v14), .macOS(.v11)],
products: [
.library(
name: "HPRTMP",
targets: ["HPRTMP"]
),
],
dependencies: [
// Dependencies declare other packages that this package depends on.
// .package(url: /* package url */, from: "1.0.0"),
.package(url: "https://github.com/apple/swift-nio", from: "2.0.0"),
],
targets: [
// Targets are the basic building blocks of a package. A target can define a module or a test suite.
// Targets can depend on other targets in this package, and on products in packages this package depends on.
.target(
name: "HPRTMP",
dependencies: []),
dependencies: [
.product(name: "NIO", package: "swift-nio")
]),
.testTarget(
name: "HPRTMPTests",
dependencies: ["HPRTMP"]),
Expand Down
120 changes: 120 additions & 0 deletions Sources/HPRTMP/Network/NetworkClient.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
import Foundation
import NIO
import os

protocol NetworkConnectable {
func connect(host: String, port: Int) async throws
func sendData(_ data: Data) async throws
func receiveData() async throws -> Data
func close() async throws
}

final class RTMPClientHandler: ChannelInboundHandler {
typealias InboundIn = ByteBuffer
private let responseCallback: (Data) -> Void

init(responseCallback: @escaping (Data) -> Void) {
self.responseCallback = responseCallback
}

func channelRead(context: ChannelHandlerContext, data: NIOAny) {
var buffer = self.unwrapInboundIn(data)

var data = Data()
buffer.readWithUnsafeReadableBytes { ptr in
data.append(ptr.baseAddress!.assumingMemoryBound(to: UInt8.self), count: ptr.count)
return ptr.count
}

guard !data.isEmpty else { return }
responseCallback(data)
}

func errorCaught(context: ChannelHandlerContext, error: Error) {
print("error: ", error)
context.close(promise: nil)
}
}

class NetworkClient: NetworkConnectable {
private let group: EventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1)
private var channel: Channel?
private var host: String?
private var port: Int?

private var cachedReceivedData: Data = .init()
private var dataPromise: EventLoopPromise<Data>?

private let logger = Logger(subsystem: "HPRTMP", category: "NetworkClient")


init() {
}

deinit {
let group = group
Task {
try? await group.shutdownGracefully()
}
}

func connect(host: String, port: Int) async throws {
self.host = host
self.port = port

let bootstrap = ClientBootstrap(group: group)
.channelInitializer { channel in
channel.pipeline.addHandlers([
RTMPClientHandler(responseCallback: self.responseReceived)
])
}

do {
self.channel = try await bootstrap.connect(host: host, port: Int(port)).get()
logger.info("[HPRTMP] Connected to \(host):\(port)")
} catch {
logger.error("[HPRTMP] Failed to connect: \(error)")
throw error
}
}

func sendData(_ data: Data) async throws {
guard let channel = self.channel else {
throw NSError(domain: "RTMPClientError", code: -1, userInfo: [NSLocalizedDescriptionKey: "Connection not established"])
}

let buffer = channel.allocator.buffer(bytes: data)
try await channel.writeAndFlush(buffer)
}

func receiveData() async throws -> Data {
guard let channel = self.channel else {
throw NSError(domain: "RTMPClientError", code: -1, userInfo: [NSLocalizedDescriptionKey: "Connection not established"])
}
if !cachedReceivedData.isEmpty {
let data = cachedReceivedData
cachedReceivedData = Data()
return data
}

self.dataPromise = channel.eventLoop.makePromise(of: Data.self)
return try await dataPromise!.futureResult.get()
}

private func responseReceived(data: Data) {
cachedReceivedData.append(data)
if let dataPromise {
dataPromise.succeed(cachedReceivedData)
cachedReceivedData = Data()
self.dataPromise = nil
}
}

func close() async throws {
dataPromise?.fail(NSError(domain: "RTMPClientError", code: -2, userInfo: [NSLocalizedDescriptionKey: "Connection invalidated"]))
let channel = self.channel
dataPromise = nil
self.channel = nil
try await channel?.close()
}
}
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review:

Bug Risks:

  1. Uninitialized DataPromise:
    • dataPromise is nil when attempting to unwrap it in receiveData().
    • Solution: Initialize dataPromise in the init() method of NetworkClient.

Improvement Suggestions:

  1. Error Handling:

    • Improve error handling by providing more specific errors and handling various failure scenarios comprehensively.

  2. Documentation:

    • Add code comments to explain complex logic, especially in DataDecoder and DataEncoder classes.
    • Document public methods, describing parameters, return values, and potential errors.

  3. Testing:

    • Implement unit tests for critical functionality like sending/receiving data and connecting/disconnecting from the server.

  4. Input Validation:

    • Validate input parameters like host, port, and data to prevent unexpected behavior or security vulnerabilities.

  5. Dependency Injection:

    • Consider injecting dependencies (like EventLoopGroup) instead of directly creating them inside classes to improve testability and flexibility.

  6. Resource Cleanup:

    • Ensure proper resource cleanup, especially in cases where exceptions can be thrown during connection or data transfer.

  7. Logging:

    • Use a logging library to handle errors and status messages consistently and make debugging easier.

  8. Error Handling Consistency:

    • Ensure consistency in error handling approaches throughout the codebase.

  9. Code Readability:

    • Improve naming conventions, add white spaces, and adhere to Swift style guidelines for better readability.

Overall, the code demonstrates a basic structure for an RTMP client. Addressing the noted issues can enhance its reliability, maintainability, and robustness.

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review:

  1. Potential Bugs/Issues:

    • In DataDecoder, the buffer.readWithUnsafeReadableBytes closure should return the number of bytes consumed, but it currently returns ptr.count. This might lead to issues if not all bytes are consumed.
    • In NetworkClient, the method responseReceived has a syntax error in the if statement condition (if let dataPromise {). It should be if let dataPromise = self.dataPromise {.
    • Error handling in sendData and receiveData methods can be improved by providing more specific error messages instead of generic ones.

  2. Improvement Suggestions:

    • Add proper error handling mechanisms using Swift's Error type throughout the codebase.
    • Consider adding timeout mechanisms when waiting for responses in the receiveData method to prevent potential deadlocks.
    • Utilize logging frameworks like os_log for better log management rather than using print statements.
    • Implement proper cleanup on unexpected errors or disconnections to release resources.

  3. General Comments:

    • The codebase appears to follow good design patterns for network communication using NIO.
    • Naming conventions are generally clear and adherent to Swift best practices.
    • More comments/documentation could improve the readability and maintainability of the code, especially for complex logic.

Overall, the code seems well-structured for networking tasks using NIO. Addressing the highlighted issues and implementing the improvement suggestions can enhance the robustness and maintainability of the codebase.

41 changes: 11 additions & 30 deletions Sources/HPRTMP/RTMPSocket.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
//

import Foundation
import Network
import os

public enum RTMPStatus {
Expand Down Expand Up @@ -67,7 +66,7 @@ protocol RTMPSocketDelegate: Actor {

public actor RTMPSocket {

private var connection: NWConnection?
private let connection: NetworkConnectable = NetworkClient()

private var status: RTMPStatus = .none

Expand Down Expand Up @@ -127,36 +126,20 @@ extension RTMPSocket {
public func resume() async {
guard status != .connected else { return }
guard let urlInfo else { return }

let port = NWEndpoint.Port(rawValue: UInt16(urlInfo.port))
let host = NWEndpoint.Host(urlInfo.host)
let connection = NWConnection(host: host, port: port ?? 1935, using: .tcp)
self.connection = connection
connection.stateUpdateHandler = { [weak self]newState in
guard let self else { return }
do {
try await connection.connect(host: urlInfo.host, port: urlInfo.port)
status = .open
Task {
switch newState {
case .ready:
self.logger.info("connection state: ready")
guard await self.status == .open else { return }
await self.startShakeHands()
case .failed(let error):
self.logger.error("[HPRTMP] connection error: \(error.localizedDescription)")
await self.delegate?.socketError(self, err: .uknown(desc: error.localizedDescription))
await self.invalidate()
default:
self.logger.info("connection state: other")
break
}
await self.startShakeHands()
}
} catch {
self.logger.error("[HPRTMP] connection error: \(error.localizedDescription)")
await self.delegate?.socketError(self, err: .uknown(desc: error.localizedDescription))
await self.invalidate()
}
NWConnection.maxReadSize = Int((await windowControl.windowSize))
status = .open
connection.start(queue: DispatchQueue.global(qos: .default))
}

private func startShakeHands() async {
guard let connection = connection else { return }
self.handshake = RTMPHandshake(dataSender: connection.sendData(_:), dataReceiver: receiveData)
await self.handshake?.setDelegate(delegate: self)
do {
Expand All @@ -173,8 +156,7 @@ extension RTMPSocket {
}
await handshake?.reset()
await decoder.reset()
connection?.cancel()
connection = nil
try? await connection.close()
urlInfo = nil
status = .closed
await delegate?.socketDisconnected(self)
Expand Down Expand Up @@ -262,15 +244,14 @@ extension RTMPSocket {

extension RTMPSocket {
private func sendData(_ data: Data) async throws {
try await connection?.sendData(data)
try await connection.sendData(data)
await windowControl.addOutBytesCount(UInt32(data.count))
}
func send(message: RTMPMessage, firstType: Bool) async {
await messagePriorityQueue.enqueue(message, firstType: firstType)
}

private func receiveData() async throws -> Data {
guard let connection = self.connection else { return Data() }
return try await connection.receiveData()
}
}
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review Summary:

Bug Risks:

  1. Connection Handling:
    • Issue: Accessing connection without checking for nil.
    • Suggestion: Use guard statements or optional chaining to safely unwrap connection.

Improvement Suggestions:

  1. Resource Management:

    • Consider using defer to ensure resources are properly released even in error scenarios.

  2. Error Handling:

    • Propagate errors from lower levels for better debugging.
    • Log more detailed error information to aid in troubleshooting.

  3. Naming and Clarity:

    • Clarify the naming conventions for variables like connection, urlInfo, etc., for better readability.

  4. Code Simplification:

    • Remove redundant or commented-out code sections to improve code cleanliness.

  5. Access Control:

    • Review access control levels; ensure that properties and methods have appropriate visibility (private, internal, public).

  6. Threading Considerations:

    • Ensure proper synchronization mechanisms if there might be thread safety concerns within asynchronous tasks.

  7. Documentation:

    • Add clear comments when necessary to explain the purpose or behavior of critical sections of the code.

  8. Testing:

    • Implement robust testing strategies to cover various edge cases and failure scenarios.

Detailed Changes:

  1. Guard against accessing connection when nil before using it in methods like receiveData() and sendData(_:).
  2. Consider implementing a defer block to guarantee proper cleanup and handling of resources.
  3. Enhance error handling by providing detailed error messages and possibly propagating errors for better upstream detection.
  4. Ensure consistency in variable naming conventions for clarity and maintainability.
  5. Remove unnecessary or obsolete code snippets to simplify the overall codebase.
  6. Verify the appropriateness of access control levels to avoid unintended access to sensitive components.
  7. Assess thread safety concerns and implement synchronization mechanisms as necessary.
  8. Document complex or critical logic segments to offer insights into their functionality.

By addressing these areas, you can enhance the reliability, readability, and maintainability of the codebase.

Expand Down
Loading