From 3f52eefb17299f6ea9e002d0751d976b3dcdfb65 Mon Sep 17 00:00:00 2001
From: Peter <iakovlev.peter.p@gmail.com>
Date: Thu, 3 Sep 2015 19:57:35 +0300
Subject: [PATCH] no message

---
 SwiftSignalKit/Queue.swift        |  8 ++++++--
 SwiftSignalKit/Signal_Catch.swift |  4 ++--
 SwiftSignalKit/Signal_Take.swift  | 14 ++++++++++++++
 SwiftSignalKit/ThreadPool.swift   | 18 +++++++++++++++---
 4 files changed, 37 insertions(+), 7 deletions(-)

diff --git a/SwiftSignalKit/Queue.swift b/SwiftSignalKit/Queue.swift
index 8408822ba5..fa46930ae6 100644
--- a/SwiftSignalKit/Queue.swift
+++ b/SwiftSignalKit/Queue.swift
@@ -38,8 +38,12 @@ public final class Queue {
         self.specialIsMainQueue = specialIsMainQueue
     }
     
-    public init() {
-        self.nativeQueue = dispatch_queue_create(nil, DISPATCH_QUEUE_SERIAL)
+    public init(name: String? = nil) {
+        if let name = name {
+            self.nativeQueue = dispatch_queue_create(name, DISPATCH_QUEUE_SERIAL)
+        } else {
+            self.nativeQueue = dispatch_queue_create(nil, DISPATCH_QUEUE_SERIAL)
+        }
         self.specific = nil
         self.specialIsMainQueue = false
         
diff --git a/SwiftSignalKit/Signal_Catch.swift b/SwiftSignalKit/Signal_Catch.swift
index f84789e844..80733ba4e6 100644
--- a/SwiftSignalKit/Signal_Catch.swift
+++ b/SwiftSignalKit/Signal_Catch.swift
@@ -90,7 +90,7 @@ public func recurse<T, E>(latestValue: T?)(signal: Signal<T, E>) -> Signal<T, E>
     }
 }
 
-public func retry<T, E>(exponentialDecay: Double, onQueue queue: Queue)(signal: Signal<T, E>) -> Signal<T, E> {
+public func retry<T, E>(delayIncrement: Double, maxDelay: Double, onQueue queue: Queue)(signal: Signal<T, E>) -> Signal<T, NoError> {
     return Signal { subscriber in
         let shouldRetry = Atomic(value: true)
         let currentDelay = Atomic(value: 0.0)
@@ -105,7 +105,7 @@ public func retry<T, E>(exponentialDecay: Double, onQueue queue: Queue)(signal:
                     subscriber.putNext(next)
                 }, error: { error in
                     let delay = currentDelay.modify { value in
-                        return value
+                        return min(maxDelay, value + delayIncrement)
                     }
                     
                     let delayTime = dispatch_time(DISPATCH_TIME_NOW, Int64(delay * Double(NSEC_PER_SEC)))
diff --git a/SwiftSignalKit/Signal_Take.swift b/SwiftSignalKit/Signal_Take.swift
index d2e11c1a2f..a7a9b563c4 100644
--- a/SwiftSignalKit/Signal_Take.swift
+++ b/SwiftSignalKit/Signal_Take.swift
@@ -25,3 +25,17 @@ public func take<T, E>(count: Int)(signal: Signal<T, E>) -> Signal<T, E> {
         })
     }
 }
+
+public func last<T, E>(signal: Signal<T, E>) -> Signal<T?, E> {
+    return Signal { subscriber in
+        let value = Atomic<T?>(value: nil)
+        return signal.start(next: { next in
+            value.swap(next)
+        }, error: { error in
+            subscriber.putError(error)
+        }, completed: { completed in
+            subscriber.putNext(value.with({ $0 }))
+            subscriber.putCompletion()
+        })
+    }
+}
diff --git a/SwiftSignalKit/ThreadPool.swift b/SwiftSignalKit/ThreadPool.swift
index acecd357aa..74b86e8c30 100644
--- a/SwiftSignalKit/ThreadPool.swift
+++ b/SwiftSignalKit/ThreadPool.swift
@@ -65,7 +65,7 @@ public func ==(lhs: ThreadPoolQueue, rhs: ThreadPoolQueue) -> Bool {
     private var mutex: pthread_mutex_t
     private var condition: pthread_cond_t
     
-    private class func threadEntryPoint(threadPool: ThreadPool) {
+    @objc class func threadEntryPoint(threadPool: ThreadPool) {
         var queue: ThreadPoolQueue!
         
         while (true) {
@@ -107,7 +107,9 @@ public func ==(lhs: ThreadPoolQueue, rhs: ThreadPoolQueue) -> Bool {
             pthread_mutex_unlock(&threadPool.mutex);
             
             if task != nil {
-                task.execute()
+                autoreleasepool {
+                    task.execute()
+                }
             }
         }
     }
@@ -123,7 +125,7 @@ public func ==(lhs: ThreadPoolQueue, rhs: ThreadPoolQueue) -> Bool {
         super.init()
         
         for _ in 0 ..< threadCount {
-            let thread = NSThread(target: ThreadPool.self, selector: Selector("threadEntryPoint"), object: self)
+            let thread = NSThread(target: ThreadPool.self, selector: Selector("threadEntryPoint:"), object: self)
             thread.threadPriority = threadPriority
             self.threads.append(thread)
             thread.start()
@@ -153,4 +155,14 @@ public func ==(lhs: ThreadPoolQueue, rhs: ThreadPoolQueue) -> Bool {
     public func nextQueue() -> ThreadPoolQueue {
         return ThreadPoolQueue(threadPool: self)
     }
+    
+    public func isCurrentThreadInPool() -> Bool {
+        let currentThread = NSThread.currentThread()
+        for thread in self.threads {
+            if currentThread.isEqual(thread) {
+                return true
+            }
+        }
+        return false
+    }
 }
\ No newline at end of file