progress

Author: zaneenders

Sources/NIOPosix/BaseStreamSocketChannel.swift

@@ -162,9 +162,23 @@ class BaseStreamSocketChannel<Socket: SocketProtocol>: BaseSocketChannel<Socket>
 
     final override func writeToSocket() throws -> OverallWriteResult {
         print(#fileID, #function)
-        // TODO: self.socket.sendmmsg(msgs: )
-        // TODO: self.socket.sendmsg(pointer: , destinationPtr: UnsafePointer<sockaddr>?, destinationSize: socklen_t, controlBytes: UnsafeMutableRawBufferPointer)
         let result = try self.pendingWrites.triggerAppropriateWriteOperations(
+            writeMessage: { ptr, destinationPtr, destinationSize, controlBytes in
+                fatalError("writeMessage")
+                guard let ctrlBytes = controlBytes else {
+                    return .processed(0)
+                }
+                var s = UnsafeControlMessageStorage.allocate(msghdrCount: 1)
+                var cb = UnsafeOutboundControlBytes(controlBytes: s[0])
+                cb.appendExplicitCongestionState(metadata: ctrlBytes, protocolFamily: .unix)
+                let controlMessageBytePointer = cb.validControlBytes
+                return try self.socket.sendmsg(
+                    pointer: ptr,
+                    destinationPtr: destinationPtr,
+                    destinationSize: destinationSize,
+                    controlBytes: controlMessageBytePointer
+                )
+            },
             scalarBufferWriteOperation: { ptr in
                 guard ptr.count > 0 else {
                     // No need to call write if the buffer is empty.

Sources/NIOPosix/PendingDatagramWritesManager.swift

@@ -365,7 +365,7 @@ private struct PendingDatagramWritesState {
         case .some(let e):
             // The compiler can't prove this, but it must be so.
             assert(self.pendingWrites.distance(from: e, to: self.pendingWrites.startIndex) == 0)
-            return .scalarBufferWrite
+            return .scalarBufferWrite(withMetaData: self.pendingWrites.first!.metadata == nil)
         default:
             return .nothingToBeWritten
         }

Sources/NIOPosix/PendingWritesManager.swift

@@ -302,7 +302,7 @@ private struct PendingStreamWritesState {
         case 1:
             switch self.pendingWrites.first!.data {
             case .byteBuffer:
-                return .scalarBufferWrite
+                return .scalarBufferWrite(withMetaData: self.pendingWrites.first!.metadata != nil)
             case .fileRegion:
                 return .scalarFileWrite
             }
@@ -315,7 +315,7 @@ private struct PendingStreamWritesState {
             case (.byteBuffer, .byteBuffer):
                 return .vectorBufferWrite
             case (.byteBuffer, .fileRegion):
-                return .scalarBufferWrite
+                return .scalarBufferWrite(withMetaData: self.pendingWrites.first!.metadata != nil)
             case (.fileRegion, _):
                 return .scalarFileWrite
             }
@@ -387,6 +387,7 @@ final class PendingStreamWritesManager: PendingWritesManager {
 
     func add(envelope: AddressedEnvelope<ByteBuffer>, promise: EventLoopPromise<Void>?) -> Bool {
         assert(self.isOpen)
+        print(#function, envelope.metadata)
         self.state.append(
             PendingStreamWrite(data: .byteBuffer(envelope.data), promise: promise, metadata: envelope.metadata)
         )
@@ -407,16 +408,22 @@ final class PendingStreamWritesManager: PendingWritesManager {
     ///   - scalarFileWriteOperation: An operation that writes a region of a file descriptor (usually `sendfile`).
     /// - Returns: The `OneWriteOperationResult` and whether the `Channel` is now writable.
     func triggerAppropriateWriteOperations(
+        writeMessage: (
+            UnsafeRawBufferPointer, UnsafePointer<sockaddr>?, socklen_t, AddressedEnvelope<ByteBuffer>.Metadata?
+        ) throws -> IOResult<Int>,
         scalarBufferWriteOperation: (UnsafeRawBufferPointer) throws -> IOResult<Int>,
         vectorBufferWriteOperation: (UnsafeBufferPointer<IOVector>) throws -> IOResult<Int>,
         scalarFileWriteOperation: (CInt, Int, Int) throws -> IOResult<Int>
     ) throws -> OverallWriteResult {
         try self.triggerWriteOperations { writeMechanism in
             print(#function, writeMechanism)
-            // TODO: add with metadata calls.
             switch writeMechanism {
-            case .scalarBufferWrite:
-                return try triggerScalarBufferWrite({ try scalarBufferWriteOperation($0) })
+            case .scalarBufferWrite(let metaData):
+                if metaData {
+                    return try _triggerScalarBufferWrite(scalarWriteOperation: { try writeMessage($0, $1, $2, $3) })
+                } else {
+                    return try triggerScalarBufferWrite({ try scalarBufferWriteOperation($0) })
+                }
             case .vectorBufferWrite:
                 return try triggerVectorBufferWrite({ try vectorBufferWriteOperation($0) })
             case .scalarFileWrite:
@@ -465,6 +472,15 @@ final class PendingStreamWritesManager: PendingWritesManager {
         }
     }
 
+    private func _triggerScalarBufferWrite(
+        scalarWriteOperation: (
+            UnsafeRawBufferPointer, UnsafePointer<sockaddr>?, socklen_t, AddressedEnvelope<ByteBuffer>.Metadata?
+        ) throws -> IOResult<Int>
+    ) rethrows -> OneWriteOperationResult {
+        fatalError("\(#function) We made it")
+        self.didWrite(itemCount: 1, result: .processed(1))
+    }
+
     /// Trigger a write of a single `FileRegion` (usually using `sendfile(2)`).
     ///
     /// - Parameters:
@@ -621,7 +637,7 @@ final class PendingStreamWritesManager: PendingWritesManager {
 }
 
 internal enum WriteMechanism {
-    case scalarBufferWrite
+    case scalarBufferWrite(withMetaData: Bool)
     case vectorBufferWrite
     case scalarFileWrite
     case nothingToBeWritten

Tests/NIOPosixTests/ChannelTests.swift

@@ -293,6 +293,9 @@ final class ChannelTests: XCTestCase {
         var multiState = 0
         var fileState = 0
         let result = try pwm.triggerAppropriateWriteOperations(
+            writeMessage: { (_, _, _, _) in
+                return .processed(0)
+            },
             scalarBufferWriteOperation: { buf in
                 defer {
                     singleState += 1