Multicast and side effects additions

SSignalKit/SMulticastSignalManager.h

@@ -5,4 +5,7 @@
 - (SSignal *)multicastedSignalForKey:(NSString *)key producer:(SSignal *(^)())producer;
 - (void)startStandaloneSignalIfNotRunningForKey:(NSString *)key producer:(SSignal *(^)())producer;
 
+- (SSignal *)multicastedPipeForKey:(NSString *)key;
+- (void)putNext:(id)next toMulticastedPipeForKey:(NSString *)key;
+
 @end

SSignalKit/SMulticastSignalManager.m

@@ -4,6 +4,7 @@
 #import "SSignal+SideEffects.h"
 #import "SBag.h"
 #import "SMetaDisposable.h"
+#import "SBlockDisposable.h"
 
 #import <libkern/OSAtomic.h>
 
@@ -12,6 +13,7 @@
     OSSpinLock _lock;
     NSMutableDictionary *_multicastSignals;
     NSMutableDictionary *_standaloneSignalDisposables;
+    NSMutableDictionary *_pipeListeners;
 }
 
 @end
@@ -25,6 +27,7 @@
     {
         _multicastSignals = [[NSMutableDictionary alloc] init];
         _standaloneSignalDisposables = [[NSMutableDictionary alloc] init];
+        _pipeListeners = [[NSMutableDictionary alloc] init];
     }
     return self;
 }
@@ -123,4 +126,42 @@
     }
 }
 
+- (SSignal *)multicastedPipeForKey:(NSString *)key
+{
+    return [[SSignal alloc] initWithGenerator:^id<SDisposable>(SSubscriber *subscriber)
+    {
+        OSSpinLockLock(&_lock);
+        SBag *bag = _pipeListeners[key];
+        if (bag == nil)
+        {
+            bag = [[SBag alloc] init];
+            _pipeListeners[key] = bag;
+        }
+        NSInteger index = [bag addItem:[^(id next)
+        {
+            [subscriber putNext:next];
+        } copy]];
+        OSSpinLockUnlock(&_lock);
+        
+        return [[SBlockDisposable alloc] initWithBlock:^
+        {
+            OSSpinLockLock(&_lock);
+            [(SBag *)_pipeListeners[key] removeItem:index];
+            OSSpinLockUnlock(&_lock);
+        }];
+    }];
+}
+
+- (void)putNext:(id)next toMulticastedPipeForKey:(NSString *)key
+{
+    OSSpinLockLock(&_lock);
+    NSArray *pipeListeners = [(SBag *)_pipeListeners[key] copyItems];
+    OSSpinLockUnlock(&_lock);
+    
+    for (void (^listener)(id) in pipeListeners)
+    {
+        listener(next);
+    }
+}
+
 @end

SSignalKit/SSignal+SideEffects.h

@@ -3,6 +3,7 @@
 @interface SSignal (SideEffects)
 
 - (SSignal *)onNext:(void (^)(id next))f;
+- (SSignal *)afterNext:(void (^)(id next))f;
 - (SSignal *)onError:(void (^)(id error))f;
 - (SSignal *)onCompletion:(void (^)())f;
 - (SSignal *)onDispose:(void (^)())f;

SSignalKit/SSignal+SideEffects.m

@@ -23,6 +23,24 @@
     }];
 }
 
+- (SSignal *)afterNext:(void (^)(id next))f
+{
+    return [[SSignal alloc] initWithGenerator:^id<SDisposable> (SSubscriber *subscriber)
+    {
+        return [self startWithNext:^(id next)
+        {
+            [subscriber putNext:next];
+            f(next);
+        } error:^(id error)
+        {
+            [subscriber putError:error];
+        } completed:^
+        {
+            [subscriber putCompletion];
+        }];
+    }];
+}
+
 - (SSignal *)onError:(void (^)(id error))f
 {
     return [[SSignal alloc] initWithGenerator:^id<SDisposable> (SSubscriber *subscriber)