Update Errors in Mqtt Client

Source/AwsCommonRuntimeKit/crt/CommonRuntimeError.swift

@@ -22,6 +22,20 @@ public struct CRTError: Equatable {
         self.message = String(cString: aws_error_str(self.code))
         self.name = String(cString: aws_error_name(self.code))
     }
+
+    public init<T: BinaryInteger>(code: T, context: String?) {
+        if code > INT32_MAX || code <= 0 {
+            self.code = Int32(AWS_ERROR_UNKNOWN.rawValue)
+        } else {
+            self.code = Int32(code)
+        }
+        var message = String(cString: aws_error_str(self.code))
+        if let context {
+            message += ": " + context
+        }
+        self.message = message
+        self.name = String(cString: aws_error_name(self.code))
+    }
 
     public static func makeFromLastError() -> CRTError {
         return CRTError(code: aws_last_error())

Source/AwsCommonRuntimeKit/crt/Lock.swift

@@ -11,15 +11,15 @@ class ReadWriteLock {
         pthread_rwlock_destroy(&rwlock)
     }
 
-    func read(_ closure: () -> Void) {
+    func read<Result>(_ closure: () throws -> Result) rethrows -> Result {
         pthread_rwlock_rdlock(&rwlock)
         defer { pthread_rwlock_unlock(&rwlock) }
-        return closure()
+        return try closure()
     }
 
-    func write( _ closure: () -> Void) {
+    func write<Result>( _ closure: () throws -> Result) rethrows -> Result {
         pthread_rwlock_wrlock(&rwlock)
         defer { pthread_rwlock_unlock(&rwlock) }
-        closure()
+        return try closure()
     }
 }

Source/AwsCommonRuntimeKit/crt/Utilities.swift

@@ -109,7 +109,7 @@
 
 extension aws_byte_buf {
     func toString() -> String {
         return String(
             data: toData(),
             encoding: .utf8)!
     }
@@ -191,7 +191,7 @@
     func toOptionalString() -> String? {
         if self.len == 0 { return nil }
         let data = Data(bytesNoCopy: self.ptr, count: self.len, deallocator: .none)
         return String(data: data, encoding: .utf8)
     }
 }
 
@@ -223,36 +223,14 @@
     }
 }
 
-/// Convert a native aws_byte_cursor pointer into a String?
-func convertAwsByteCursorToOptionalString(_ awsByteCursor: UnsafePointer<aws_byte_cursor>?) -> String? {
-    guard let cursor = awsByteCursor?.pointee else {
-        return nil
-    }
-    return cursor.toString()
-}
-
-/// Convert a native uint16_t pointer into a Swift UInt16?
-func convertOptionalUInt16(_ pointer: UnsafePointer<UInt16>?) -> UInt16? {
-    guard let validPointer = pointer else {
-        return nil
-    }
-    return validPointer.pointee
-}
-
-/// Convert a native uint32_t pointer into a Swift UInt32?
-func convertOptionalUInt32(_ pointer: UnsafePointer<UInt32>?) -> UInt32? {
-    guard let validPointer = pointer else {
-        return nil
-    }
-    return validPointer.pointee
-}
-
-/// Convert a native bool pointer to an optional Swift Bool
-func convertOptionalBool(_ pointer: UnsafePointer<Bool>?) -> Bool? {
-    guard let validPointer = pointer else {
-        return nil
+extension Optional {
+    /// unwrap an optional unsafepointer to its underlying type
+    func unwrap<T>() -> T? where Wrapped == (UnsafePointer<T>) {
+        guard let validPointer = self else {
+            return nil
+        }
+        return validPointer.pointee
     }
-    return validPointer.pointee
 }
 
 extension Bool {

Source/AwsCommonRuntimeKit/mqtt/Mqtt5Enums.swift

@@ -3,11 +3,6 @@
 
 import AwsCMqtt
 
-// TODO this is temporary. We will replace this with aws-crt-swift error codes.
-enum MqttError: Error {
-    case validation(message: String)
-}
-
 /// MQTT message delivery quality of service.
 /// Enum values match `MQTT5 spec <https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901234>`__ encoding values.
 public enum QoS {
@@ -595,4 +590,4 @@ extension InboundTopicAliasBehaviorType {
         case .disabled: return AWS_MQTT5_CITABT_DISABLED
         }
     }
-}
+}

Source/AwsCommonRuntimeKit/mqtt/Mqtt5Options.swift

@@ -110,20 +110,23 @@
     func validateConversionToNative() throws {
         if let keepAliveInterval {
             if keepAliveInterval < 0 || keepAliveInterval > Double(UInt16.max) {
-                throw MqttError.validation(message: "Invalid keepAliveInterval value")
+                throw CommonRunTimeError.crtError(CRTError(code: AWS_ERROR_INVALID_ARGUMENT.rawValue,
+                                                           context: "Invalid keepAliveInterval value"))
             }
         }
 
         do {
             _ = try sessionExpiryInterval?.secondUInt32()
         } catch {
-            throw MqttError.validation(message: "Invalid sessionExpiryInterval value")
+            throw CommonRunTimeError.crtError(CRTError(code: AWS_ERROR_INVALID_ARGUMENT.rawValue,
+                                                       context: "Invalid sessionExpiryInterval value"))
         }
 
         do {
             _ = try willDelayInterval?.secondUInt32()
         } catch {
-            throw MqttError.validation(message: "Invalid willDelayInterval value")
+            throw CommonRunTimeError.crtError(CRTError(code: AWS_ERROR_INVALID_ARGUMENT.rawValue,
+                                                       context: "Invalid willDelayInterval value"))
         }
     }
 
@@ -148,11 +151,11 @@
             _willDelayIntervalSec,
             self.receiveMaximum,
             self.maximumPacketSize) { sessionExpiryIntervalSecPointer,
                                       requestResponseInformationPointer,
                                       requestProblemInformationPointer,
                                       willDelayIntervalSecPointer,
                                       receiveMaximumPointer,
                                       maximumPacketSizePointer in
 
                 raw_connect_options.session_expiry_interval_seconds = sessionExpiryIntervalSecPointer
                 raw_connect_options.request_response_information = requestResponseInformationPointer
@@ -339,36 +342,42 @@
         do {
             _ = try minReconnectDelay?.millisecondUInt64()
         } catch {
-            throw MqttError.validation(message: "Invalid minReconnectDelay value")
+            throw CommonRunTimeError.crtError(CRTError(code: AWS_ERROR_INVALID_ARGUMENT.rawValue,
+                                                       context: "Invalid minReconnectDelay value"))
         }
 
         do {
             _ = try maxReconnectDelay?.millisecondUInt64()
         } catch {
-            throw MqttError.validation(message: "Invalid maxReconnectDelay value")
+            throw CommonRunTimeError.crtError(CRTError(code: AWS_ERROR_INVALID_ARGUMENT.rawValue,
+                                                       context: "Invalid maxReconnectDelay value"))
         }
 
         do {
             _ = try minConnectedTimeToResetReconnectDelay?.millisecondUInt64()
         } catch {
-            throw MqttError.validation(message: "Invalid minConnectedTimeToResetReconnectDelay value")
+            throw CommonRunTimeError.crtError(CRTError(code: AWS_ERROR_INVALID_ARGUMENT.rawValue,
+                                                       context: "Invalid minConnectedTimeToResetReconnectDelay value"))
         }
 
         do {
             _ = try pingTimeout?.millisecondUInt32()
         } catch {
-            throw MqttError.validation(message: "Invalid pingTimeout value")
+            throw CommonRunTimeError.crtError(CRTError(code: AWS_ERROR_INVALID_ARGUMENT.rawValue,
+                                                       context: "Invalid pingTimeout value"))
         }
 
         do {
             _ = try connackTimeout?.millisecondUInt32()
         } catch {
-            throw MqttError.validation(message: "Invalid connackTimeout value")
+            throw CommonRunTimeError.crtError(CRTError(code: AWS_ERROR_INVALID_ARGUMENT.rawValue,
+                                                       context: "Invalid connackTimeout value"))
         }
 
         if let ackTimeout {
             if ackTimeout < 0 || ackTimeout > Double(UInt32.max) {
-                throw MqttError.validation(message: "Invalid ackTimeout value")
+                throw CommonRunTimeError.crtError(CRTError(code: AWS_ERROR_INVALID_ARGUMENT.rawValue,
+                                                           context: "Invalid ackTimeout value"))
             }
         }
     }
@@ -438,7 +447,7 @@
             self.httpProxyOptions,
             self.topicAliasingOptions,
             connnectOptions) { socketOptionsCPointer,
                                tlsOptionsCPointer,
                                httpProxyOptionsCPointer,
                                topicAliasingOptionsCPointer,
                                connectOptionsCPointer in
@@ -449,7 +458,7 @@
                 raw_options.topic_aliasing_options = topicAliasingOptionsCPointer
                 raw_options.connect_options = connectOptionsCPointer
 
-                guard let _userData = userData else {
+                guard let userData else {
                     // directly return
                     return hostName.withByteCursor { hostNameByteCursor in
                         raw_options.host_name = hostNameByteCursor
@@ -459,15 +468,15 @@
 
                 if self.onWebsocketTransform != nil {
                     raw_options.websocket_handshake_transform = MqttClientWebsocketTransform
-                    raw_options.websocket_handshake_transform_user_data = _userData
+                    raw_options.websocket_handshake_transform_user_data = userData
                 }
 
                 raw_options.lifecycle_event_handler = MqttClientHandleLifecycleEvent
-                raw_options.lifecycle_event_handler_user_data = _userData
+                raw_options.lifecycle_event_handler_user_data = userData
                 raw_options.publish_received_handler = MqttClientHandlePublishRecieved
-                raw_options.publish_received_handler_user_data = _userData
+                raw_options.publish_received_handler_user_data = userData
                 raw_options.client_termination_handler = MqttClientTerminationCallback
-                raw_options.client_termination_handler_user_data = _userData
+                raw_options.client_termination_handler_user_data = userData
                 return hostName.withByteCursor { hostNameByteCursor in
                     raw_options.host_name = hostNameByteCursor
                     return body(raw_options)
@@ -524,7 +533,7 @@
     /// The final client id in use by the newly-established connection.  This will be the configured client id if one was given in the configuration, otherwise, if no client id was specified, this will be the client id assigned by the server.  Reconnection attempts will always use the auto-assigned client id, allowing for auto-assigned session resumption.
     public let clientId: String
 
-    internal init(_ settings: UnsafePointer<aws_mqtt5_negotiated_settings>){
+    internal init(_ settings: UnsafePointer<aws_mqtt5_negotiated_settings>) {
         let negotiatedSettings = settings.pointee
         self.maximumQos = QoS(negotiatedSettings.maximum_qos)
         self.sessionExpiryInterval = TimeInterval(negotiatedSettings.session_expiry_interval)

Source/AwsCommonRuntimeKit/mqtt/Mqtt5Packets.swift

@@ -176,9 +176,9 @@
         self.retain = publishView.retain
         self.payloadFormatIndicator = publishView.payload_format != nil ?
             PayloadFormatIndicator(publishView.payload_format.pointee) : nil
-        self.messageExpiryInterval = convertOptionalUInt32(publishView.message_expiry_interval_seconds).map { TimeInterval($0) }
-        self.topicAlias = convertOptionalUInt16(publishView.topic_alias)
-        self.responseTopic = convertAwsByteCursorToOptionalString(publishView.response_topic)
+        self.messageExpiryInterval = publishView.message_expiry_interval_seconds.unwrap().map { TimeInterval($0) }
+        self.topicAlias = publishView.topic_alias.unwrap()
+        self.responseTopic = publishView.response_topic?.pointee.toString()
         self.correlationData = publishView.correlation_data != nil ?
             Data(bytes: publishView.correlation_data!.pointee.ptr, count: publishView.correlation_data!.pointee.len) : nil
         var identifier: [UInt32]? = []
@@ -187,16 +187,16 @@
             identifier?.append(subscription_identifier)
         }
         self.subscriptionIdentifiers = identifier
-        self.contentType = convertAwsByteCursorToOptionalString(publishView.content_type)
+        self.contentType = publishView.content_type?.pointee.toString()
         self.userProperties = convertOptionalUserProperties(
             count: publishView.user_property_count,
             userPropertiesPointer: publishView.user_properties)
     }
 
     /// Get payload converted to a utf8 String
     public func payloadAsString() -> String? {
-        if let payload{
+        if let payload {
             return String(data: payload, encoding: .utf8) ?? nil
         }
         return nil
     }
@@ -204,7 +204,8 @@
     func validateConversionToNative() throws {
         if let messageExpiryInterval {
             if messageExpiryInterval < 0 || messageExpiryInterval > Double(UInt32.max) {
-                throw MqttError.validation(message: "Invalid sessionExpiryInterval value")
+                throw CommonRunTimeError.crtError(CRTError(code: AWS_ERROR_INVALID_ARGUMENT.rawValue, 
+                                                           context: "Invalid sessionExpiryInterval value"))
             }
         }
     }
@@ -460,7 +461,7 @@
                         to: self.subscriptionIdentifier) { identiferPointer in
                             raw_subscrbe_view.subscription_identifier = identiferPointer
                             return body(raw_subscrbe_view)
-                        }
+                    }
             }
         }
     }
@@ -582,7 +583,7 @@
         self.userProperties = userProperties
     }
 
-    internal init(_ unsuback_view: UnsafePointer<aws_mqtt5_packet_unsuback_view>){
+    internal init(_ unsuback_view: UnsafePointer<aws_mqtt5_packet_unsuback_view>) {
         let unsubackView = unsuback_view.pointee
         let reasonCodeBuffer = UnsafeBufferPointer(start: unsubackView.reason_codes, count: unsubackView.reason_code_count)
         self.reasonCodes = reasonCodeBuffer.compactMap { UnsubackReasonCode(rawValue: Int($0.rawValue)) }
@@ -623,13 +624,13 @@
             self.userProperties = userProperties
     }
 
-    internal init(_ disconnect_view: UnsafePointer<aws_mqtt5_packet_disconnect_view>){
+    internal init(_ disconnect_view: UnsafePointer<aws_mqtt5_packet_disconnect_view>) {
         let disconnectView = disconnect_view.pointee
 
         self.reasonCode = DisconnectReasonCode(rawValue: Int(disconnectView.reason_code.rawValue))!
-        self.sessionExpiryInterval = convertOptionalUInt32(disconnectView.session_expiry_interval_seconds).map { TimeInterval($0) }
-        self.reasonString = convertAwsByteCursorToOptionalString(disconnectView.reason_string)
-        self.serverReference = convertAwsByteCursorToOptionalString(disconnectView.reason_string)
+        self.sessionExpiryInterval = disconnectView.session_expiry_interval_seconds.unwrap().map { TimeInterval($0) }
+        self.reasonString = disconnectView.reason_string?.pointee.toString()
+        self.serverReference = disconnectView.reason_string?.pointee.toString()
         self.userProperties = convertOptionalUserProperties(
             count: disconnectView.user_property_count,
             userPropertiesPointer: disconnectView.user_properties)
@@ -638,7 +639,8 @@
     func validateConversionToNative() throws {
         if let sessionExpiryInterval {
             if sessionExpiryInterval < 0 || sessionExpiryInterval > Double(UInt32.max) {
-                throw MqttError.validation(message: "Invalid sessionExpiryInterval value")
+                throw CommonRunTimeError.crtError(CRTError(code: AWS_ERROR_INVALID_ARGUMENT.rawValue,
+                                                           context: "Invalid sessionExpiryInterval value"))
             }
         }
     }
@@ -768,29 +770,29 @@
         self.serverReference = serverReference
     }
 
-    internal init(_ connack_view: UnsafePointer<aws_mqtt5_packet_connack_view>){
+    internal init(_ connack_view: UnsafePointer<aws_mqtt5_packet_connack_view>) {
         let connackView = connack_view.pointee
 
         self.sessionPresent = connackView.session_present
         self.reasonCode = ConnectReasonCode(rawValue: Int(connackView.reason_code.rawValue))!
         self.sessionExpiryInterval = (connackView.session_expiry_interval?.pointee).map { TimeInterval($0) }
-        self.receiveMaximum = convertOptionalUInt16(connackView.receive_maximum)
+        self.receiveMaximum = connackView.receive_maximum.unwrap()
         if let maximumQosValue = connackView.maximum_qos {
             self.maximumQos = QoS(maximumQosValue.pointee)
         } else {
             self.maximumQos = nil
         }
-        self.retainAvailable = convertOptionalBool(connackView.retain_available)
-        self.maximumPacketSize = convertOptionalUInt32(connackView.maximum_packet_size)
-        self.assignedClientIdentifier = convertAwsByteCursorToOptionalString(connackView.assigned_client_identifier)
-        self.topicAliasMaximum = convertOptionalUInt16(connackView.topic_alias_maximum)
-        self.reasonString = convertAwsByteCursorToOptionalString(connackView.reason_string)
-        self.wildcardSubscriptionsAvailable = convertOptionalBool(connackView.wildcard_subscriptions_available)
-        self.subscriptionIdentifiersAvailable = convertOptionalBool(connackView.subscription_identifiers_available)
-        self.sharedSubscriptionAvailable = convertOptionalBool(connackView.shared_subscriptions_available)
-        self.serverKeepAlive = convertOptionalUInt16(connackView.server_keep_alive).map { TimeInterval($0) }
-        self.responseInformation = convertAwsByteCursorToOptionalString(connackView.response_information)
-        self.serverReference = convertAwsByteCursorToOptionalString(connackView.server_reference)
+        self.retainAvailable = connackView.retain_available.unwrap()
+        self.maximumPacketSize = connackView.maximum_packet_size.unwrap()
+        self.assignedClientIdentifier = connackView.assigned_client_identifier?.pointee.toString()
+        self.topicAliasMaximum = connackView.topic_alias_maximum.unwrap()
+        self.reasonString = connackView.reason_string?.pointee.toString()
+        self.wildcardSubscriptionsAvailable = connackView.wildcard_subscriptions_available.unwrap()
+        self.subscriptionIdentifiersAvailable = connackView.subscription_identifiers_available.unwrap()
+        self.sharedSubscriptionAvailable = connackView.shared_subscriptions_available.unwrap()
+        self.serverKeepAlive = connackView.server_keep_alive.unwrap().map { TimeInterval($0) }
+        self.responseInformation = connackView.response_information?.pointee.toString()
+        self.serverReference = connackView.server_reference?.pointee.toString()
         self.userProperties = convertOptionalUserProperties(
             count: connackView.user_property_count,
             userPropertiesPointer: connackView.user_properties)