SQueue fixes

reduceLeftWithPassthrough

SSignalKit/SQueue.m

@@ -80,7 +80,12 @@ static const void *SQueueSpecificKey = &SQueueSpecificKey;
 
 - (void)dispatch:(dispatch_block_t)block
 {
-    dispatch_async(_queue, block);
+    if (_queueSpecific != NULL && dispatch_get_specific(SQueueSpecificKey) == _queueSpecific)
+        block();
+    else if (_specialIsMainQueue && [NSThread isMainThread])
+        block();
+    else
+        dispatch_async(_queue, block);
 }
 
 - (void)dispatchSync:(dispatch_block_t)block

SSignalKit/SSignal+Accumulate.h

@@ -3,5 +3,6 @@
 @interface SSignal (Accumulate)
 
 - (SSignal *)reduceLeft:(id)value with:(id (^)(id, id))f;
+- (SSignal *)reduceLeftWithPassthrough:(id)value with:(id (^)(id, id, void (^)(id)))f;
 
 @end

SSignalKit/SSignal+Accumulate.m

@@ -23,4 +23,30 @@
     }];
 }
 
+- (SSignal *)reduceLeftWithPassthrough:(id)value with:(id (^)(id, id, void (^)(id)))f
+{
+    return [[SSignal alloc] initWithGenerator:^(SSubscriber *subscriber)
+    {
+        __block id intermediateResult = value;
+        
+        void (^emit)(id) = ^(id next)
+        {
+            [subscriber putNext:next];
+        };
+        
+        return [self startWithNext:^(id next)
+        {
+            intermediateResult = f(intermediateResult, next, emit);
+        } error:^(id error)
+        {
+            [subscriber putError:error];
+        } completed:^
+        {
+            if (intermediateResult != nil)
+                [subscriber putNext:intermediateResult];
+            [subscriber putCompletion];
+        }];
+    }];
+}
+
 @end