From ef3aa40c2df28721f83ae695047b6cc23ef3753e Mon Sep 17 00:00:00 2001
From: Srdan Rasic <srdan.rasic@gmail.com>
Date: Sat, 22 Jun 2019 09:35:31 +0200
Subject: [PATCH] Add sink operator.

---
 Sources/SignalProtocol.swift | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/Sources/SignalProtocol.swift b/Sources/SignalProtocol.swift
index 7052ecb..30e2a04 100644
--- a/Sources/SignalProtocol.swift
+++ b/Sources/SignalProtocol.swift
@@ -80,3 +80,20 @@ extension SignalProtocol {
         return (self as? Signal<Element, Error>) ?? Signal(self.observe)
     }
 }
+
+extension SignalProtocol {
+
+    /// Attaches a subscriber (observer) with closure-based behavior.
+    public func sink(receiveCompletion: ((Subscribers.Completion<Error>) -> Void)? = nil, receiveValue: @escaping ((Element) -> Void)) -> Disposable {
+        return observe { event in
+            switch event {
+            case .next(let element):
+                receiveValue(element)
+            case .failed(let error):
+                receiveCompletion?(.failure(error))
+            case .completed:
+                receiveCompletion?(.finished)
+            }
+        }
+    }
+}