diff --git a/examples/DelegateObservableExample.swift b/examples/DelegateObservableExample.swift
index a727de9..a11def4 100644
--- a/examples/DelegateObservableExample.swift
+++ b/examples/DelegateObservableExample.swift
@@ -67,20 +67,20 @@ public class DelegateObservableExampleViewController: UIViewController {
     let pan = UIPanGestureRecognizer()
     view.addGestureRecognizer(pan)
 
-    let dragStream = IndefiniteObservable<DragProducer.Value> { observer in
+    let dragStream = IndefiniteObservable { observer in
       return DragProducer(subscribedTo: pan, observer: observer).unsubscribe
     }
 
     // Must hold a reference to the subscription, otherwise the stream will be deallocated when the
     // subscription goes out of scope.
-    subscriptions.append(dragStream.subscribe {
+    subscriptions.append(dragStream.subscribe(observer: ValueObserver {
       if $0.state == .began || $0.state == .changed {
         targetView.layer.position = $0.location
       }
-    })
+    }))
 
-    subscriptions.append(dragStream.subscribe {
+    subscriptions.append(dragStream.subscribe(observer: ValueObserver {
       print($0.state.rawValue)
-    })
+    }))
   }
 }
diff --git a/examples/OperatorExample.swift b/examples/OperatorExample.swift
index 51e0891..525b80d 100644
--- a/examples/OperatorExample.swift
+++ b/examples/OperatorExample.swift
@@ -17,14 +17,30 @@
 import UIKit
 import IndefiniteObservable
 
-// This example demonstrates how to create custom operators that can be chained to an
-// IndefiniteObservable.
+// This example demonstrates how to create a custom observable/observer type and to add operators to
+// it.
 
-extension IndefiniteObservable {
+public final class ValueObserver<T> {
+  public typealias Value = T
+
+  public init(_ next: @escaping (T) -> Void) {
+    self.next = next
+  }
+
+  public let next: (T) -> Void
+}
+
+public class ValueObservable<T>: IndefiniteObservable<ValueObserver<T>> {
+  public final func subscribe(_ next: @escaping (T) -> Void) -> Subscription {
+    return super.subscribe(observer: ValueObserver(next))
+  }
+}
+
+extension ValueObservable {
 
   // Map from one value type to another.
-  public func map<U>(_ transform: @escaping (T) -> U) -> IndefiniteObservable<U> {
-    return IndefiniteObservable<U> { observer in
+  public func map<U>(_ transform: @escaping (T) -> U) -> ValueObservable<U> {
+    return ValueObservable<U> { observer in
       return self.subscribe {
         observer.next(transform($0))
       }.unsubscribe
@@ -32,8 +48,8 @@ extension IndefiniteObservable {
   }
 
   // Only emit values downstream for which passesTest returns true
-  public func filter(_ passesTest: @escaping (T) -> Bool) -> IndefiniteObservable<T> {
-    return IndefiniteObservable<T> { observer in
+  public func filter(_ passesTest: @escaping (T) -> Bool) -> ValueObservable<T> {
+    return ValueObservable<T> { observer in
       return self.subscribe {
         if passesTest($0) {
           observer.next($0)
@@ -43,6 +59,43 @@ extension IndefiniteObservable {
   }
 }
 
+public enum MotionState {
+  case atRest
+  case active
+}
+
+public final class MotionObserver<T> {
+  public typealias Value = T
+
+  public init(next: @escaping (T) -> Void, state: @escaping (MotionState) -> Void) {
+    self.next = next
+    self.state = state
+  }
+
+  public let next: (T) -> Void
+  public let state: (MotionState) -> Void
+}
+
+public class MotionObservable<T>: IndefiniteObservable<MotionObserver<T>> {
+  public final func subscribe(next: @escaping (T) -> Void, state: @escaping (MotionState) -> Void) -> Subscription {
+    return super.subscribe(observer: MotionObserver(next: next, state: state))
+  }
+}
+
+extension MotionObservable {
+
+  // Map from one value type to another.
+  public func map<U>(_ transform: @escaping (T) -> U) -> MotionObservable<U> {
+    return MotionObservable<U> { observer in
+      return self.subscribe(next: {
+        observer.next(transform($0))
+      }, state: { state in
+        observer.state(state)
+      }).unsubscribe
+    }
+  }
+}
+
 public class OperatorExampleViewController: UIViewController {
 
   var initialPosition: CGPoint = .zero
@@ -59,11 +112,17 @@ public class OperatorExampleViewController: UIViewController {
     let pan = UIPanGestureRecognizer()
     view.addGestureRecognizer(pan)
 
-    let dragStream = IndefiniteObservable<DragProducer.Value> { observer in
+    let dragStream = ValueObservable<DragProducer.Value> { observer in
       return DragProducer(subscribedTo: pan, observer: observer).unsubscribe
     }
 
-    // Note that we avoid keep a strong reference to self in the stream's operators.
+    let motionStream = MotionObservable<Int> { observer in
+      observer.next(5)
+      observer.state(.atRest)
+      return noopUnsubscription
+    }
+
+    // Note that we avoid keeping a strong reference to self in the stream's operators.
     // A strong reference would create a retain cycle:
     //
     // subscription -> stream -> operator -> self -> subscriptions
@@ -75,9 +134,9 @@ public class OperatorExampleViewController: UIViewController {
       .filter { $0.state == .began || $0.state == .changed }
       .map { $0.location }
       .map { .init(x: midX, y: $0.y) }
-      .subscribe {
+      .subscribe(observer: ValueObserver {
         targetView.layer.position = $0
-      }
+      })
     )
   }
 }
diff --git a/src/IndefiniteObservable.swift b/src/IndefiniteObservable.swift
index b988251..8f8bea8 100644
--- a/src/IndefiniteObservable.swift
+++ b/src/IndefiniteObservable.swift
@@ -46,11 +46,11 @@
        }
      }
  */
-public final class IndefiniteObservable<T> {
-  public typealias Subscriber<T> = (ValueObserver<T>) -> (() -> Void)?
+open class IndefiniteObservable<O> {
+  public typealias Subscriber<O> = (O) -> (() -> Void)?
 
   /** A subscriber is only invoked when subscribe is invoked. */
-  public init(_ subscriber: @escaping Subscriber<T>) {
+  public init(_ subscriber: @escaping Subscriber<O>) {
     self.subscriber = subscriber
   }
 
@@ -65,8 +65,7 @@ public final class IndefiniteObservable<T> {
    - Parameter next: A block that will be executed when new values are sent from upstream.
    - Returns: A subscription.
    */
-  public func subscribe(next: @escaping (T) -> Void) -> Subscription {
-    let observer = ValueObserver<T>(next)
+  public final func subscribe(observer: O) -> Subscription {
     if let subscription = subscriber(observer) {
       return SimpleSubscription(subscription)
     } else {
@@ -74,7 +73,7 @@ public final class IndefiniteObservable<T> {
     }
   }
 
-  private let subscriber: Subscriber<T>
+  private let subscriber: Subscriber<O>
 }
 
 /** A Subscription is returned by IndefiniteObservable.subscribe. */
@@ -97,19 +96,6 @@ public protocol Subscription {
  */
 public let noopUnsubscription: (() -> Void)? = nil
 
-// MARK: Type erasing
-
-/** An ValueObserver receives data from an IndefiniteObservable. */
-public final class ValueObserver<T> {
-  public typealias Value = T
-
-  public init(_ next: @escaping (T) -> Void) {
-    self.next = next
-  }
-
-  public let next: (T) -> Void
-}
-
 // MARK: Private
 
 // Internal class for ensuring that an active subscription keeps its stream alive.
diff --git a/tests/unit/MemoryLeakTests.swift b/tests/unit/MemoryLeakTests.swift
index 13c5df0..cf6eded 100644
--- a/tests/unit/MemoryLeakTests.swift
+++ b/tests/unit/MemoryLeakTests.swift
@@ -20,7 +20,7 @@ import IndefiniteObservable
 
 class MemoryLeakTests: XCTestCase {
   func testObservableIsDeallocated() {
-    var observable: IndefiniteObservable<CGFloat>? = IndefiniteObservable<CGFloat> { observer in
+    var observable: ValueObservable<CGFloat>? = ValueObservable<CGFloat> { observer in
       observer.next(5)
       return noopUnsubscription
     }
@@ -37,7 +37,7 @@ class MemoryLeakTests: XCTestCase {
   }
 
   func testDownstreamObservableKeepsUpstreamAlive() {
-    var observable: IndefiniteObservable<CGFloat>? = IndefiniteObservable<CGFloat> { observer in
+    var observable: ValueObservable<CGFloat>? = ValueObservable<CGFloat> { observer in
       observer.next(5)
       return noopUnsubscription
     }
@@ -56,7 +56,7 @@ class MemoryLeakTests: XCTestCase {
   }
 
   func testSubscribedObservableIsDeallocated() {
-    var observable: IndefiniteObservable<CGFloat>? = IndefiniteObservable<CGFloat> { observer in
+    var observable: ValueObservable<CGFloat>? = ValueObservable<CGFloat> { observer in
       observer.next(5)
       return noopUnsubscription
     }
@@ -76,7 +76,7 @@ class MemoryLeakTests: XCTestCase {
   }
 
   func testSubscribedObservableWithOperatorIsDeallocated() {
-    var observable: IndefiniteObservable<CGFloat>? = IndefiniteObservable<CGFloat> { observer in
+    var observable: ValueObservable<CGFloat>? = ValueObservable<CGFloat> { observer in
       observer.next(5)
       return noopUnsubscription
     }
@@ -98,9 +98,9 @@ class MemoryLeakTests: XCTestCase {
   }
 
   func testUnsubscribedObservableWithOperatorIsDeallocated() {
-    weak var weakObservable: IndefiniteObservable<CGFloat>?
+    weak var weakObservable: ValueObservable<CGFloat>?
     autoreleasepool {
-      let observable: IndefiniteObservable<CGFloat>? = IndefiniteObservable<CGFloat> { observer in
+      let observable: ValueObservable<CGFloat>? = ValueObservable<CGFloat> { observer in
         observer.next(5)
         return noopUnsubscription
       }
@@ -121,12 +121,12 @@ class MemoryLeakTests: XCTestCase {
   }
 
   func testSubscriptionDoesNotKeepObservableInMemory() {
-    weak var weakObservable: IndefiniteObservable<Int>?
+    weak var weakObservable: ValueObservable<Int>?
     var subscription: Subscription?
 
     autoreleasepool {
       let value = 10
-      let observable = IndefiniteObservable<Int> { observer in
+      let observable = ValueObservable<Int> { observer in
         observer.next(value)
         return noopUnsubscription
       }
diff --git a/tests/unit/ObservableTests.swift b/tests/unit/ObservableTests.swift
index ca2ce57..434cee2 100644
--- a/tests/unit/ObservableTests.swift
+++ b/tests/unit/ObservableTests.swift
@@ -23,7 +23,7 @@ class ObservableTests: XCTestCase {
   func testSubscription() {
     let value = 10
 
-    let observable = IndefiniteObservable<Int> { observer in
+    let observable = ValueObservable<Int> { observer in
       observer.next(value)
       return noopUnsubscription
     }
@@ -42,7 +42,7 @@ class ObservableTests: XCTestCase {
     var didUnsubscribe = false
 
     autoreleasepool {
-      let observable = IndefiniteObservable<CGFloat> { observer in
+      let observable = ValueObservable<CGFloat> { observer in
         return {
           didUnsubscribe = true
         }
@@ -57,7 +57,7 @@ class ObservableTests: XCTestCase {
   func testUnsubscribesOnUnsubscribe() {
     var didUnsubscribe = false
 
-    let observable = IndefiniteObservable<CGFloat> { observer in
+    let observable = ValueObservable<CGFloat> { observer in
       return {
         didUnsubscribe = true
       }
@@ -72,7 +72,7 @@ class ObservableTests: XCTestCase {
   func testTwoSubsequentSubscriptions() {
     let value = 10
 
-    let observable = IndefiniteObservable<Int> { observer in
+    let observable = ValueObservable<Int> { observer in
       observer.next(value)
       return noopUnsubscription
     }
@@ -97,7 +97,7 @@ class ObservableTests: XCTestCase {
   func testTwoParalellSubscriptions() {
     let value = 10
 
-    let observable = IndefiniteObservable<Int> { observer in
+    let observable = ValueObservable<Int> { observer in
       observer.next(value)
       return noopUnsubscription
     }
@@ -124,7 +124,7 @@ class ObservableTests: XCTestCase {
 
   func testMappingValues() {
     let value = 10
-    let observable = IndefiniteObservable<Int> { observer in
+    let observable = ValueObservable<Int> { observer in
       observer.next(value)
       return noopUnsubscription
     }
@@ -141,7 +141,7 @@ class ObservableTests: XCTestCase {
 
   func testMappingTypes() {
     let value = CGPoint(x: 0, y: 10)
-    let observable = IndefiniteObservable<CGPoint> { observer in
+    let observable = ValueObservable<CGPoint> { observer in
       observer.next(value)
       return noopUnsubscription
     }
@@ -158,7 +158,7 @@ class ObservableTests: XCTestCase {
 
   func testFilteringValues() {
     let value = CGPoint(x: 0, y: 10)
-    let observable = IndefiniteObservable<(Bool, CGPoint)> { observer in
+    let observable = ValueObservable<(Bool, CGPoint)> { observer in
       observer.next(false, value)
       observer.next(true, value)
       return noopUnsubscription
@@ -194,7 +194,7 @@ class ObservableTests: XCTestCase {
   func testGeneratedValuesAreReceived() {
     let generator = DeferredGenerator()
 
-    let observable = IndefiniteObservable<Int> { observer in
+    let observable = ValueObservable<Int> { observer in
       generator.addObserver(observer)
       return {
         generator.removeObserver(observer)
@@ -223,7 +223,7 @@ class ObservableTests: XCTestCase {
   func testGeneratedValuesAreNotReceivedAfterUnsubscription() {
     let generator = DeferredGenerator()
 
-    let observable = IndefiniteObservable<Int> { observer in
+    let observable = ValueObservable<Int> { observer in
       generator.addObserver(observer)
       return {
         generator.removeObserver(observer)
@@ -250,11 +250,11 @@ class ObservableTests: XCTestCase {
   }
 
   func testGeneratedValuesAreNotReceivedAfterUnsubscriptionOrder2() {
-    weak var weakObservable: IndefiniteObservable<Int>?
+    weak var weakObservable: ValueObservable<Int>?
     autoreleasepool {
       let generator = DeferredGenerator()
 
-      let observable = IndefiniteObservable<Int> { observer in
+      let observable = ValueObservable<Int> { observer in
         generator.addObserver(observer)
         return {
           generator.removeObserver(observer)
diff --git a/tests/unit/SimpleOperators.swift b/tests/unit/SimpleOperators.swift
index 7e2d70c..4b0e971 100644
--- a/tests/unit/SimpleOperators.swift
+++ b/tests/unit/SimpleOperators.swift
@@ -19,25 +19,41 @@ import IndefiniteObservable
 
 // Simple operators used by the tests.
 
-extension IndefiniteObservable {
+public final class ValueObserver<T> {
+  public typealias Value = T
+
+  public init(_ next: @escaping (T) -> Void) {
+    self.next = next
+  }
+
+  public let next: (T) -> Void
+}
+
+public class ValueObservable<T>: IndefiniteObservable<ValueObserver<T>> {
+  public final func subscribe(_ next: @escaping (T) -> Void) -> Subscription {
+    return super.subscribe(observer: ValueObserver(next))
+  }
+}
+
+extension ValueObservable {
 
   // Map from one value type to another.
-  public func map<U>(_ transform: @escaping (T) -> U) -> IndefiniteObservable<U> {
-    return IndefiniteObservable<U> { observer in
-      return self.subscribe {
+  public func map<U>(_ transform: @escaping (T) -> U) -> ValueObservable<U> {
+    return ValueObservable<U> { observer in
+      return self.subscribe(observer: ValueObserver<T> {
         observer.next(transform($0))
-      }.unsubscribe
+      }).unsubscribe
     }
   }
 
   // Only emit values downstream for which passesTest returns true
-  public func filter(_ passesTest: @escaping (T) -> Bool) -> IndefiniteObservable<T> {
-    return IndefiniteObservable<T> { observer in
-      return self.subscribe {
+  public func filter(_ passesTest: @escaping (T) -> Bool) -> ValueObservable<T> {
+    return ValueObservable<T> { observer in
+      return self.subscribe(observer: ValueObserver<T> {
         if passesTest($0) {
           observer.next($0)
         }
-      }.unsubscribe
+      }).unsubscribe
     }
   }
 }