Remove warnings

Flow/Callbacker.swift

@@ -21,20 +21,22 @@ public final class Callbacker<Value> {
 
     private var callbacks = Callbacks.none
     private var _mutex = pthread_mutex_t()
-    private var mutex: PThreadMutex { return PThreadMutex(&_mutex) }
+    private func withMutex<T>(_ body: (PThreadMutex) throws -> T) rethrows -> T {
+        try withUnsafeMutablePointer(to: &_mutex, body)
+    }
 
     public init() {
-        mutex.initialize()
+        withMutex { $0.initialize() }
     }
 
     deinit {
-        mutex.deinitialize()
+        withMutex { $0.deinitialize() }
     }
 
     /// - Returns: True if no callbacks has been registered.
     public var isEmpty: Bool {
-        mutex.lock()
-        defer { mutex.unlock() }
+        withMutex { $0.lock() }
+        defer { withMutex { $0.unlock() } }
 
         switch callbacks {
         case .none: return true
@@ -46,8 +48,8 @@ public final class Callbacker<Value> {
     /// Register a callback to be called when `callAll` is executed.
     /// - Returns: A `Disposable` to be disposed to unregister the callback.
     public func addCallback(_ callback: @escaping (Value) -> Void) -> Disposable {
-        mutex.lock()
-        defer { mutex.unlock() }
+        withMutex { $0.lock() }
+        defer { withMutex { $0.unlock() } }
 
         let key = generateKey()
 
@@ -63,8 +65,8 @@ public final class Callbacker<Value> {
         }
 
         return NoLockKeyDisposer(key) { key in
-            self.mutex.lock()
-            defer { self.mutex.unlock() }
+            self.withMutex { $0.lock() }
+            defer { self.withMutex { $0.unlock() } }
 
             switch self.callbacks {
             case .single(let singleKey, _) where singleKey == key:
@@ -82,9 +84,9 @@ public final class Callbacker<Value> {
 
     /// Will call all registered callbacks with `value`
     public func callAll(with value: Value) {
-        mutex.lock()
+        withMutex { $0.lock() }
         let callbacks = self.callbacks
-        mutex.unlock()
+        withMutex { $0.unlock() }
 
         switch callbacks {
         case .none: break

Flow/Disposable.swift

@@ -29,24 +29,26 @@ public struct NilDisposer: Disposable {
 public final class Disposer: Disposable {
     private var disposer: (() -> ())?
     private var _mutex = pthread_mutex_t()
-    private var mutex: PThreadMutex { return PThreadMutex(&_mutex) }
+    private func withMutex<T>(_ body: (PThreadMutex) throws -> T) rethrows -> T {
+        try withUnsafeMutablePointer(to: &_mutex, body)
+    }
 
     /// Pass a closure to be called when being disposed
     public init(_ disposer: @escaping () -> () = {}) {
         self.disposer = disposer
-        mutex.initialize()
+        withMutex { $0.initialize() }
     }
 
     deinit {
         dispose()
-        mutex.deinitialize()
+        withMutex { $0.deinitialize() }
     }
 
     public func dispose() {
-        mutex.lock()
+        withMutex { $0.lock() }
         let disposer = self.disposer
         self.disposer = nil
-        mutex.unlock()
+        withMutex { $0.unlock() }
         disposer?()
     }
 }
@@ -59,48 +61,50 @@ public final class Disposer: Disposable {
 public final class DisposeBag: Disposable {
     private var disposables: [Disposable]
     private var _mutex = pthread_mutex_t()
-    private var mutex: PThreadMutex { return PThreadMutex(&_mutex) }
+    private func withMutex<T>(_ body: (PThreadMutex) throws -> T) rethrows -> T {
+        try withUnsafeMutablePointer(to: &_mutex, body)
+    }
 
     /// Create an empty instance
     public init() {
         self.disposables = []
-        mutex.initialize()
+        withMutex { $0.initialize() }
     }
 
     /// Create an instance already containing `disposables`
     public init<S: Sequence>(_ disposables: S) where S.Iterator.Element == Disposable {
         self.disposables = Array(disposables)
-        mutex.initialize()
+        withMutex { $0.initialize() }
     }
 
     /// Create an instance already containing `disposables`
     public init(_ disposables: Disposable...) {
         self.disposables = disposables
-        mutex.initialize()
+        withMutex { $0.initialize() }
     }
 
     deinit {
         dispose()
-        mutex.deinitialize()
+        withMutex { $0.deinitialize() }
     }
 
     /// Returns true if there is currently no disposables to dispose.
     public var isEmpty: Bool {
-        return mutex.protect { disposables.isEmpty }
+        return withMutex { $0.protect { disposables.isEmpty } }
     }
 
     public func dispose() {
-        mutex.lock()
+        withMutex { $0.lock() }
         let disposables = self.disposables // make sure to make a copy in the case any call to dispose will recursivaly call us back.
         self.disposables = []
-        mutex.unlock()
+        withMutex { $0.unlock() }
         for disposable in disposables { disposable.dispose() }
     }
 
     /// Add `disposable` to `self`
     public func add(_ disposable: Disposable) {
-        mutex.lock()
-        defer { mutex.unlock() }
+        withMutex { $0.lock() }
+        defer { withMutex { $0.unlock() } }
         disposables.append(disposable)
     }
 }

Flow/Future+Combiners.swift

@@ -65,9 +65,9 @@ public func join<T>(_ futures: [Future<T>], cancelNonCompleted: Bool = true) ->
     var results = [T?](repeating: nil, count: futures.count)
     let mutex = Mutex()
     func onValue(_ i: Int, _ val: T) {
-        mutex.protect {
-            results[i] = val
-        }
+        mutex.lock()
+        results[i] = val
+        mutex.unlock()
     }
 
     var future = futures.first!.onValue(on: .none) { onValue(0, $0) }
@@ -220,7 +220,9 @@ public final class SingleTaskPerformer<Value> {
 
         mutex.unlock() // unlock while calling out as we might either recurs or always might execute at once.
         let singleFuture = function().always(on: .none) {
-            self.mutex.protect { self.future = nil }
+            self.mutex.lock()
+            self.future = nil
+            self.mutex.unlock()
         }
         mutex.lock()
 
@@ -233,7 +235,9 @@ public final class SingleTaskPerformer<Value> {
     }
 
     public var isPerforming: Bool {
-        return mutex.protect { self.future != nil }
+        mutex.lock()
+        defer { mutex.unlock() }
+        return self.future != nil
     }
 }
 

Flow/Future.swift

@@ -107,7 +107,7 @@ public final class Future<Value> {
                 try onResult(completion, Mover(shouldClone: true))
             }
         }
-        mutex.initialize()
+        withMutex { $0.initialize() }
 
         scheduler.async {
             do {
@@ -143,13 +143,13 @@ public final class Future<Value> {
 
         state = .completed(result)
         clone = { Future(result: result) }
-        mutex.initialize()
+        withMutex { $0.initialize() }
     }
 
     deinit {
         OSAtomicDecrement32(&futureUnitTestAliveCount)
         memPrint("Future deinit", futureUnitTestAliveCount)
-        mutex.deinitialize()
+        withMutex { $0.deinitialize() }
     }
 }
 
@@ -327,18 +327,20 @@ func memPrint(_ str: String, _ count: Int32) {
 }
 
 private extension Future {
-    var mutex: PThreadMutex { return PThreadMutex(&_mutex) }
+    private func withMutex<T>(_ body: (PThreadMutex) throws -> T) rethrows -> T {
+        try withUnsafeMutablePointer(to: &_mutex, body)
+    }
 
     private var protectedState: State {
-        return mutex.protect { state }
+        return withMutex { $0.protect { state } }
     }
 
     func lock() {
-        mutex.lock()
+        withMutex { $0.lock() }
     }
 
     func unlock() {
-        mutex.unlock()
+        withMutex { $0.unlock() }
     }
 
     func completeWithResult(_ result: Result<Value>) {

Flow/FutureQueue.swift

@@ -61,9 +61,9 @@ public extension FutureQueue {
         return Future { completion in
             let item = QueueItem<Output>(operation: operation, completion: completion)
 
-            self.mutex.protect {
-                self.items.append(item)
-            }
+            self.withMutex { $0.lock() }
+            self.items.append(item)
+            self.withMutex { $0.unlock() }
 
             self.executeNextItem()
 
@@ -119,7 +119,7 @@ public extension FutureQueue {
 public extension FutureQueue {
     /// Do we have any enqueued operations?
     var isEmpty: Bool {
-        return mutex.protect { items.isEmpty }
+        return withMutex { $0.protect { items.isEmpty } }
     }
 
     /// Returns a signal that will signal when `isEmpty` is changed.
@@ -164,19 +164,22 @@ public extension FutureQueue {
 
     /// The error passed to `abortQueuedExecutionWithError()` if called with `shouldCloseQueue` as true.
     var closedError: Error? {
-        return mutex.protect { _closedError }
+        return withMutex { $0.protect { _closedError } }
     }
 }
 
 private extension FutureQueue {
-    var mutex: PThreadMutex { return PThreadMutex(&_mutex) }
-    func lock() { mutex.lock() }
-    func unlock() { mutex.unlock() }
+    private func withMutex<T>(_ body: (PThreadMutex) throws -> T) rethrows -> T {
+        try withUnsafeMutablePointer(to: &_mutex, body)
+    }
+
+    func lock() { withMutex { $0.lock() } }
+    func unlock() { withMutex { $0.unlock() } }
 
     func removeItem(_ item: Executable) {
-        mutex.protect {
-            _ = items.firstIndex { $0 === item }.map { items.remove(at: $0) }
-        }
+        withMutex { $0.lock() }
+        _ = items.firstIndex { $0 === item }.map { items.remove(at: $0) }
+        withMutex { $0.unlock() }
     }
 
     func executeNextItem() {
@@ -188,9 +191,9 @@ private extension FutureQueue {
         unlock()
 
         item.execute(on: queueScheduler) {
-            self.mutex.protect {
-                self.concurrentCount -= 1
-            }
+            self.lock()
+            self.concurrentCount -= 1
+            self.unlock()
             self.removeItem(item)
             self.executeNextItem()
         }
@@ -215,25 +218,27 @@ private final class QueueItem<Output>: Executable {
     private weak var future: Future<Output>?
     private var hasBeenCancelled = false
     private var _mutex = pthread_mutex_t()
+    private func withMutex<T>(_ body: (PThreadMutex) throws -> T) rethrows -> T {
+        try withUnsafeMutablePointer(to: &_mutex, body)
+    }
 
     init(operation: @escaping () throws -> Future<Output>, completion: @escaping (Result<Output>) -> ()) {
         self.completion = completion
         self.operation = operation
-        mutex.initialize()
+        withMutex { $0.initialize() }
 
         OSAtomicIncrement32(&queueItemUnitTestAliveCount)
         memPrint("Queue Item init", queueItemUnitTestAliveCount)
     }
 
     deinit {
-        mutex.deinitialize()
+        withMutex { $0.deinitialize() }
         OSAtomicDecrement32(&queueItemUnitTestAliveCount)
         memPrint("Queue Item deinit", queueItemUnitTestAliveCount)
     }
 
-    private var mutex: PThreadMutex { return PThreadMutex(&_mutex) }
-    private func lock() { mutex.lock() }
-    private func unlock() { mutex.unlock() }
+    private func lock() { withMutex { $0.lock() } }
+    private func unlock() { withMutex { $0.unlock() } }
 
     private func complete(_ result: (Result<Output>)) {
         lock()