   /// Batch each `size` elements into another signal.
   public func window(size: Int) -> Signal<Signal<Element, Error>, Error> {
     return buffer(size: size).map { Signal.sequence($0) }
+  }
+}
+
+extension SignalProtocol where Element: OptionalProtocol {
+
+  /// Apply `transform` to all non-nil elements.
+  public func flatMap<U>(_ transform: @escaping (Element.Wrapped) -> U?) -> Signal<U?, Error> {
+    return Signal { observer in
+      return self.observe { event in
+        switch event {
+        case .next(let element):
+          if let element = element._unbox {
+            observer.next(transform(element))
+          } else {
+            observer.next(nil)
+          }
+        case .failed(let error):
+          observer.failed(error)
+        case .completed:
+          observer.completed()
+        }
+      }
+    }
+  }
+}
+
+extension SignalProtocol where Element: Collection {
+
+  /// Map each emitted array.
+  public func flatMap<U>(_ transform: @escaping (Element.Iterator.Element) -> U) -> Signal<[U], Error> {
+    return Signal { observer in
+      return self.observe { event in
+        switch event {
+        case .next(let element):
+          observer.next(element.map(transform))
+        case .failed(let error):
+          observer.failed(error)
+        case .completed:
+          observer.completed()
+        }
+      }
+    }
   }
 }
 

    return castError()._with(latestFrom: other, combine: { ($0, $1) })
  }
}

      }
    }
  }

  /// Wrap events into elements.
  public func materialize() -> Signal<Event<Element, Error>, NoError> {
    return Signal { observer in
      return self.observe { event in
        switch event {
        case .next(let element):
          observer.next(.next(element))
        case .failed(let error):
          observer.next(.failed(error))
          observer.completed()
        case .completed:
          observer.next(.completed)
          observer.completed()
        }
      }
    }
  }
}

// MARK: Injections
