diff --git a/.jazzy.yaml b/.jazzy.yaml
index b4411b5..ef940d4 100644
--- a/.jazzy.yaml
+++ b/.jazzy.yaml
@@ -1,5 +1,5 @@
 module: IndefiniteObservable
-module_version: 1.0.0
+module_version: 2.0.0
 sdk: iphonesimulator
 xcodebuild_arguments:
   - -workspace
@@ -7,4 +7,4 @@ xcodebuild_arguments:
   - -scheme
   - IndefiniteObservable
 github_url: https://github.com/material-motion/indefinite-observable-swift
-github_file_prefix: https://github.com/material-motion/indefinite-observable-swift/tree/v1.0.0
+github_file_prefix: https://github.com/material-motion/indefinite-observable-swift/tree/v2.0.0
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 17ed46d..4aa702e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,72 @@
+# 2.0.0
+
+IndefiniteObservable now supports novel Observer implementations using a genericized Observer type.
+
+## Breaking changes
+
+- Removed `AnyObserver`.
+- Renamed `noUnsubscription` to `noopUnsubscription`.
+
+## New features
+
+- `IndefiniteObservable` now requires a generic Observer type, enabling the creation of custom
+  observers with channels other than `next`.
+
+## Source changes
+
+* [Make the shape of the default Observer have a value and next function.](https://github.com/material-motion/indefinite-observable-swift/commit/a78e255c60b2ba855d16333216a78ed6147389ac) (Jeff Verkoeyen)
+* [Make IndefiniteObservable observer-agnostic.](https://github.com/material-motion/indefinite-observable-swift/commit/8c30e9a2517211c6d7a36b9c5e129d2dd3ee5dda) (Jeff Verkoeyen)
+* [Cleaned up and added some additional unit tests](https://github.com/material-motion/indefinite-observable-swift/commit/0a630845641602724853ccce7279331adb87368f) (Jeff Verkoeyen)
+* [Remove unnecessary UpstreamSubscription.](https://github.com/material-motion/indefinite-observable-swift/commit/0914e5d866f30f756f9d5799cac6d55572190567) (Jeff Verkoeyen)
+* [Set next in the ValueObserver wrather than wrap it.](https://github.com/material-motion/indefinite-observable-swift/commit/ec6d4b1410ee46ce0a46854fb14bfb9ddc0074fa) (Jeff Verkoeyen)
+* [Make IndefiniteObservable final so that it can't be subclassed.](https://github.com/material-motion/indefinite-observable-swift/commit/fd5869de6aeb69add9db5f820064f604cb4d8e78) (Jeff Verkoeyen)
+* [Rename noUnsubscription to noopUnsubscription.](https://github.com/material-motion/indefinite-observable-swift/commit/7ff94398942bdcbc91230330e4541ae8e5bef296) (Jeff Verkoeyen)
+* [Remove the unused Observer type.](https://github.com/material-motion/indefinite-observable-swift/commit/bad9ca6f410d5ff6d9ee9b3fb142b7c120e15b6a) (Jeff Verkoeyen)
+* [Rename AnyObserver to ValueObserver.](https://github.com/material-motion/indefinite-observable-swift/commit/b6a280e41ee9410e5e14c9d3368fb24b60ad1577) (Jeff Verkoeyen)
+
+## API changes
+
+Auto-generated by running:
+
+    apidiff origin/stable release-candidate swift IndefiniteObservable.xcworkspace IndefiniteObservable
+
+## noUnsubscription
+
+*renamed* global var: `noUnsubscription` → `noopUnsubscription`
+
+## Observer
+
+*new* var: `next` in `Observer`
+
+*removed* method: `next(_:)` in `Observer`
+
+## AnyObserver
+
+*removed* class: `AnyObserver`
+
+*removed* method: `next(_:)` in `AnyObserver`
+
+## IndefiniteObservable
+
+*new* method: `subscribe(observer:)` in `IndefiniteObservable`
+
+*removed* method: `subscribe(next:)` in `IndefiniteObservable`
+
+*modified* class: `IndefiniteObservable`
+
+| Type of change: | key.doc.declaration |
+|---|---|
+| From: | `open class IndefiniteObservable<T>` |
+| To: | `open class IndefiniteObservable<O : Observer>` |
+
+## Non-source changes
+
+* [Automatic changelog preparation for release.](https://github.com/material-motion/indefinite-observable-swift/commit/f1f6e215013af5927dcda4e6a305714df6745732) (Jeff Verkoeyen)
+* [Better podspec description.](https://github.com/material-motion/indefinite-observable-swift/commit/11df6b0e78930daed9503f5b119d59e4d4316d2f) (Jeff Verkoeyen)
+* [Fix typo in README.md guides.](https://github.com/material-motion/indefinite-observable-swift/commit/8f762cdea31b76d9fb1ccb77547a4ffbdc585e12) (Jeff Verkoeyen)
+* [Undo the previous change.](https://github.com/material-motion/indefinite-observable-swift/commit/ad148e61004f6aaf55527765067fe72297e1d362) (Jeff Verkoeyen)
+* [Poke the travis build.](https://github.com/material-motion/indefinite-observable-swift/commit/6e0168c3f2f475c8a60aee7e8173d56af738cdfe) (Jeff Verkoeyen)
+
 # 1.0.0
 
 Stable release of `IndefiniteObservable`.
diff --git a/IndefiniteObservable.podspec b/IndefiniteObservable.podspec
index 0e08284..c55b008 100644
--- a/IndefiniteObservable.podspec
+++ b/IndefiniteObservable.podspec
@@ -1,7 +1,7 @@
 Pod::Spec.new do |s|
   s.name         = "IndefiniteObservable"
-  s.summary      = "IndefiniteObservable.swift"
-  s.version      = "1.0.0"
+  s.summary      = "IndefiniteObservable is a minimal implementation of Observable with no concept of completion or failure."
+  s.version      = "2.0.0"
   s.authors      = "The Material Motion Authors"
   s.license      = "Apache 2.0"
   s.homepage     = "https://github.com/material-motion/indefinite-observable-swift"
diff --git a/Podfile.lock b/Podfile.lock
index 047d050..e0916ca 100644
--- a/Podfile.lock
+++ b/Podfile.lock
@@ -1,8 +1,8 @@
 PODS:
   - CatalogByConvention (2.0.0)
-  - IndefiniteObservable/examples (1.0.0):
+  - IndefiniteObservable/examples (2.0.0):
     - IndefiniteObservable/lib
-  - IndefiniteObservable/lib (1.0.0)
+  - IndefiniteObservable/lib (2.0.0)
 
 DEPENDENCIES:
   - CatalogByConvention
@@ -15,7 +15,7 @@ EXTERNAL SOURCES:
 
 SPEC CHECKSUMS:
   CatalogByConvention: be55c2263132e4f9f59299ac8a528ee8715b3275
-  IndefiniteObservable: b90809f2fa025e37124c14681192bc441b128625
+  IndefiniteObservable: 66776cf0f5929a5a260bf04d340bfb0a7087d56f
 
 PODFILE CHECKSUM: 3e4cdba95901e07a289159f0c5a8b830ecb1a5c8
 
diff --git a/README.md b/README.md
index 08a0bdf..7c178d9 100644
--- a/README.md
+++ b/README.md
@@ -54,18 +54,42 @@ commands:
 
 # Guides
 
+1. [How to make an observable](#how-to-make-an-observable)
 1. [How to create a synchronous stream](#how-to-create-a-synchronous-stream)
-2. [How to create an asynchronous stream using blocks](#how-to-create-an-asynchronous-stream-using-blocks)
-3. [How to subscribe to a stream](#how-to-subscribe-to-a-stream)
-4. [How to unsubscribe from a stream](#how-to-unsubscribe-from-a-stream)
-5. [How to create an synchronous stream using objects](#how-to-create-an-synchronous-stream-using-objects)
+1. [How to create an asynchronous stream using blocks](#how-to-create-an-asynchronous-stream-using-blocks)
+1. [How to subscribe to a stream](#how-to-subscribe-to-a-stream)
+1. [How to unsubscribe from a stream](#how-to-unsubscribe-from-a-stream)
+1. [How to create an synchronous stream using objects](#how-to-create-an-synchronous-stream-using-objects)
+
+## How to make an observable
+
+In this example we'll make the simplest possible observable type: a value observable. We will use
+this concrete type in all of the following guides.
+
+```swift
+final class ValueObserver<T>: Observer {
+  typealias Value = T
+
+  init(_ next: @escaping (T) -> Void) {
+    self.next = next
+  }
+
+  let next: (T) -> Void
+}
+
+final class ValueObservable<T>: IndefiniteObservable<ValueObserver<T>> {
+  func subscribe(_ next: @escaping (T) -> Void) -> Subscription {
+    return super.subscribe(observer: ValueObserver(next))
+  }
+}
+```
 
 ## How to create a synchronous stream
 
 ```swift
-let observable = IndefiniteObservable<<#ValueType#>> { observer in
+let observable = ValueObservable<<#ValueType#>> { observer in
   observer.next(<#value#>)
-  return noUnsubscription
+  return noopUnsubscription
 }
 ```
 
@@ -75,7 +99,7 @@ If you have an API that provides a block-based mechanism for registering observe
 create an asynchronous stream in place like so:
 
 ```swift
-let observable = IndefiniteObservable<<#ValueType#>> { observer in
+let observable = ValueObservable<<#ValueType#>> { observer in
   let someToken = registerSomeCallback { callbackValue in
     observer.next(callbackValue)
   }
@@ -117,7 +141,7 @@ need to create a `Producer` class. A `Producer` listens for events with an event
 class DragProducer: Subscription {
   typealias Value = (state: UIGestureRecognizerState, location: CGPoint)
 
-  init(subscribedTo gesture: UIPanGestureRecognizer, observer: AnyObserver<Value>) {
+  init(subscribedTo gesture: UIPanGestureRecognizer, observer: ValueObserver<Value>) {
     self.gesture = gesture
     self.observer = observer
 
@@ -141,14 +165,14 @@ class DragProducer: Subscription {
   }
 
   var gesture: (UIPanGestureRecognizer)?
-  let observer: AnyObserver<Value>
+  let observer: ValueObserver<Value>
 }
 
 let pan = UIPanGestureRecognizer()
 view.addGestureRecognizer(pan)
 
-let dragStream = IndefiniteObservable<DragProducer.Value> { observer in
-  return DragProducer(subscribedTo: pan, observer: observer).subscription
+let dragStream = ValueObservable<DragProducer.Value> { observer in
+  return DragProducer(subscribedTo: pan, observer: observer).unsubscribe
 }
 let subscription = dragStream.subscribe {
   dump($0.state)
@@ -175,22 +199,22 @@ class DragProducer: Subscription {
 
 ### Step 3: Implement the initializer
 
-Your initializer must accept and store an `AnyObserver<Value>` instance.
+Your initializer must accept and store an `ValueObserver<Value>` instance.
 
 ```swift
-  init(subscribedTo gesture: UIPanGestureRecognizer, observer: AnyObserver<Value>) {
+  init(subscribedTo gesture: UIPanGestureRecognizer, observer: ValueObserver<Value>) {
     self.gesture = gesture
     self.observer = observer
   }
 
   var gesture: (UIPanGestureRecognizer)?
-  let observer: AnyObserver<Value>
+  let observer: ValueObserver<Value>
 ```
 
 ### Step 4: Connect to the event source and send values to the observer
 
 ```swift
-  init(subscribedTo gesture: UIPanGestureRecognizer, observer: AnyObserver<Value>) {
+  init(subscribedTo gesture: UIPanGestureRecognizer, observer: ValueObserver<Value>) {
     ...
 
     gesture.addTarget(self, action: #selector(didPan))
@@ -221,7 +245,7 @@ You are responsible for disconnecting from and releasing any resources here.
 It often is helpful to provide the observer with the current state on registration.
 
 ```swift
-  init(subscribedTo gesture: UIPanGestureRecognizer, observer: AnyObserver<Value>) {
+  init(subscribedTo gesture: UIPanGestureRecognizer, observer: ValueObserver<Value>) {
     ...
 
     // Populate the observer with the current gesture state.
@@ -232,8 +256,8 @@ It often is helpful to provide the observer with the current state on registrati
 ### Step 7: Observe the producer
 
 ```swift
-let dragStream = IndefiniteObservable<DragProducer.Value> { observer in
-  return DragProducer(subscribedTo: pan, observer: observer).subscription
+let dragStream = ValueObservable<DragProducer.Value> { observer in
+  return DragProducer(subscribedTo: pan, observer: observer).unsubscribe
 }
 let subscription = dragStream.subscribe {
   dump($0)
diff --git a/examples/DelegateObservableExample.swift b/examples/DelegateObservableExample.swift
index 06f229a..a11def4 100644
--- a/examples/DelegateObservableExample.swift
+++ b/examples/DelegateObservableExample.swift
@@ -24,7 +24,7 @@ import IndefiniteObservable
 class DragProducer: Subscription {
   typealias Value = (state: UIGestureRecognizerState, location: CGPoint)
 
-  init(subscribedTo gesture: UIPanGestureRecognizer, observer: AnyObserver<Value>) {
+  init(subscribedTo gesture: UIPanGestureRecognizer, observer: ValueObserver<Value>) {
     self.gesture = gesture
     self.observer = observer
 
@@ -48,7 +48,7 @@ class DragProducer: Subscription {
   }
 
   var gesture: (UIPanGestureRecognizer)?
-  let observer: AnyObserver<Value>
+  let observer: ValueObserver<Value>
 }
 
 public class DelegateObservableExampleViewController: UIViewController {
@@ -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..bcd5adb 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>: Observer {
+  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>: Observer {
+  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 2a7ee35..43525fe 100644
--- a/src/IndefiniteObservable.swift
+++ b/src/IndefiniteObservable.swift
@@ -25,7 +25,7 @@
 
      let observable = IndefiniteObservable<Int> { observer in
        observer.next(5)
-       return noUnsubscription
+       return noopUnsubscription
      }
 
      let subscription = observable.subscribe { value in
@@ -46,11 +46,11 @@
        }
      }
  */
-open class IndefiniteObservable<T> {
-  public typealias Subscriber<T> = (AnyObserver<T>) -> (() -> Void)?
+open class IndefiniteObservable<O: Observer> {
+  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,26 +65,21 @@ open 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 = AnyObserver<T>(next)
-
-    // This line creates our "downstream" data flow.
-    let subscription = subscriber(AnyObserver { observer.next($0) })
-
-    // We store a strong reference to self in the subscription in order to keep the stream alive.
-    // When the subscription goes away, so does the stream.
-    return UpstreamSubscription(observable: self) {
-      subscription?()
+  public final func subscribe(observer: O) -> Subscription {
+    if let subscription = subscriber(observer) {
+      return SimpleSubscription(subscription)
+    } else {
+      return SimpleSubscription()
     }
   }
 
-  private let subscriber: Subscriber<T>
+  private let subscriber: Subscriber<O>
 }
 
-/** An Observer receives data from an IndefiniteObservable. */
+/** An Observer is provided to an Observable's subscribe method. */
 public protocol Observer {
   associatedtype Value
-  func next(_ value: Value) -> Void
+  var next: (Value) -> Void { get }
 }
 
 /** A Subscription is returned by IndefiniteObservable.subscribe. */
@@ -102,45 +97,33 @@ public protocol Subscription {
      let observable = IndefiniteObservable<Int> { observer in
        observer.next(5)
 
-       return noUnsubscription
+       return noopUnsubscription
      }
  */
-public let noUnsubscription: (() -> Void)? = nil
-
-// MARK: Type erasing
-
-/** A type-erased observer. */
-public final class AnyObserver<T>: Observer {
-  public typealias Value = T
-
-  init(_ next: @escaping (Value) -> Void) {
-    _next = next
-  }
-
-  public func next(_ value: Value) {
-    _next(value)
-  }
-
-  private let _next: (Value) -> Void
-}
+public let noopUnsubscription: (() -> Void)? = nil
 
 // MARK: Private
 
 // Internal class for ensuring that an active subscription keeps its stream alive.
 // Streams don't hold strong references down the chain, so our subscriptions hold strong references
 // "up" the chain to the IndefiniteObservable type.
-private final class UpstreamSubscription: Subscription {
-  init(observable: Any, _ unsubscribe: @escaping () -> Void) {
-    _observable = observable
+private final class SimpleSubscription: Subscription {
+  deinit {
+    unsubscribe()
+  }
+
+  init(_ unsubscribe: @escaping () -> Void) {
     _unsubscribe = unsubscribe
   }
 
+  init() {
+    _unsubscribe = nil
+  }
+
   func unsubscribe() {
     _unsubscribe?()
     _unsubscribe = nil
-    _observable = nil
   }
 
   private var _unsubscribe: (() -> Void)?
-  private var _observable: Any?
 }
diff --git a/tests/unit/MemoryLeakTests.swift b/tests/unit/MemoryLeakTests.swift
index bc9ff89..cf6eded 100644
--- a/tests/unit/MemoryLeakTests.swift
+++ b/tests/unit/MemoryLeakTests.swift
@@ -20,9 +20,9 @@ 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 noUnsubscription
+      return noopUnsubscription
     }
     weak var weakObservable = observable
 
@@ -37,9 +37,9 @@ 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 noUnsubscription
+      return noopUnsubscription
     }
     weak var weakObservable = observable
 
@@ -56,16 +56,16 @@ 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 noUnsubscription
+      return noopUnsubscription
     }
     weak var weakObservable = observable
 
     autoreleasepool {
-      let _ = observable!.subscribe(next: {
+      let _ = observable!.subscribe {
         let _ = $0
-      })
+      }
       // Remove our only strong reference.
       observable = nil
     }
@@ -76,18 +76,18 @@ 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 noUnsubscription
+      return noopUnsubscription
     }
     weak var weakObservable = observable
 
     autoreleasepool {
       let _ = observable!.map { value in
         return value * value
-        }.subscribe(next: {
+        }.subscribe {
           let _ = $0
-        })
+        }
       // Remove our only strong reference.
       observable = nil
     }
@@ -98,19 +98,19 @@ 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 noUnsubscription
+        return noopUnsubscription
       }
       weakObservable = observable
 
       let subscription = observable!.map { value in
         return value * value
-        }.subscribe(next: {
+        }.subscribe {
           let _ = $0
-        })
+        }
       // Remove our only strong reference.
       subscription.unsubscribe()
     }
@@ -120,27 +120,23 @@ class MemoryLeakTests: XCTestCase {
     XCTAssertNil(weakObservable)
   }
 
-  func testSubscriptionKeepsObservableInMemory() {
-    weak var weakObservable: IndefiniteObservable<Int>?
+  func testSubscriptionDoesNotKeepObservableInMemory() {
+    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 noUnsubscription
+        return noopUnsubscription
       }
       weakObservable = observable
 
       subscription = observable.subscribe { _ in }
     }
 
-    XCTAssertNotNil(weakObservable)
+    XCTAssertNil(weakObservable)
 
     subscription?.unsubscribe()
-
-    // If this fails it means there's a retain cycle. Place a breakpoint here and use the Debug
-    // Memory Graph tool to debug.
-    XCTAssertNil(weakObservable)
   }
 }
diff --git a/tests/unit/ObservableTests.swift b/tests/unit/ObservableTests.swift
index 8de9fe5..434cee2 100644
--- a/tests/unit/ObservableTests.swift
+++ b/tests/unit/ObservableTests.swift
@@ -23,9 +23,9 @@ class ObservableTests: XCTestCase {
   func testSubscription() {
     let value = 10
 
-    let observable = IndefiniteObservable<Int> { observer in
+    let observable = ValueObservable<Int> { observer in
       observer.next(value)
-      return noUnsubscription
+      return noopUnsubscription
     }
 
     let wasReceived = expectation(description: "Value was received")
@@ -38,12 +38,43 @@ class ObservableTests: XCTestCase {
     waitForExpectations(timeout: 0)
   }
 
+  func testUnsubscribesOnDeallocation() {
+    var didUnsubscribe = false
+
+    autoreleasepool {
+      let observable = ValueObservable<CGFloat> { observer in
+        return {
+          didUnsubscribe = true
+        }
+      }
+
+      let _ = observable.subscribe { _ in }
+    }
+
+    XCTAssertTrue(didUnsubscribe)
+  }
+
+  func testUnsubscribesOnUnsubscribe() {
+    var didUnsubscribe = false
+
+    let observable = ValueObservable<CGFloat> { observer in
+      return {
+        didUnsubscribe = true
+      }
+    }
+
+    let subscription = observable.subscribe { _ in }
+    subscription.unsubscribe()
+
+    XCTAssertTrue(didUnsubscribe)
+  }
+
   func testTwoSubsequentSubscriptions() {
     let value = 10
 
-    let observable = IndefiniteObservable<Int> { observer in
+    let observable = ValueObservable<Int> { observer in
       observer.next(value)
-      return noUnsubscription
+      return noopUnsubscription
     }
 
     let wasReceived = expectation(description: "Value was received")
@@ -66,9 +97,9 @@ class ObservableTests: XCTestCase {
   func testTwoParalellSubscriptions() {
     let value = 10
 
-    let observable = IndefiniteObservable<Int> { observer in
+    let observable = ValueObservable<Int> { observer in
       observer.next(value)
-      return noUnsubscription
+      return noopUnsubscription
     }
 
     let wasReceived = expectation(description: "Value was received")
@@ -93,9 +124,9 @@ class ObservableTests: XCTestCase {
 
   func testMappingValues() {
     let value = 10
-    let observable = IndefiniteObservable<Int> { observer in
+    let observable = ValueObservable<Int> { observer in
       observer.next(value)
-      return noUnsubscription
+      return noopUnsubscription
     }
 
     let wasReceived = expectation(description: "Value was received")
@@ -110,9 +141,9 @@ 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 noUnsubscription
+      return noopUnsubscription
     }
 
     let wasReceived = expectation(description: "Value was received")
@@ -127,10 +158,10 @@ class ObservableTests: XCTestCase {
 
   func testFilteringValues() {
     let value = CGPoint(x: 0, y: 10)
-    let observable = IndefiniteObservable<(Bool, CGPoint)> { observer in
-      observer.next((false, value))
-      observer.next((true, value))
-      return noUnsubscription
+    let observable = ValueObservable<(Bool, CGPoint)> { observer in
+      observer.next(false, value)
+      observer.next(true, value)
+      return noopUnsubscription
     }
 
     var filteredValues: [CGPoint] = []
@@ -142,11 +173,11 @@ class ObservableTests: XCTestCase {
   }
 
   class DeferredGenerator {
-    func addObserver(_ observer: AnyObserver<Int>) {
+    func addObserver(_ observer: ValueObserver<Int>) {
       observers.append(observer)
     }
 
-    func removeObserver(_ observer: AnyObserver<Int>) {
+    func removeObserver(_ observer: ValueObserver<Int>) {
       if let index = observers.index(where: { $0 === observer }) {
         observers.remove(at: index)
       }
@@ -157,13 +188,13 @@ class ObservableTests: XCTestCase {
         observer.next(value)
       }
     }
-    var observers: [AnyObserver<Int>] = []
+    var observers: [ValueObserver<Int>] = []
   }
 
   func testGeneratedValuesAreReceived() {
     let generator = DeferredGenerator()
 
-    let observable = IndefiniteObservable<Int> { observer in
+    let observable = ValueObservable<Int> { observer in
       generator.addObserver(observer)
       return {
         generator.removeObserver(observer)
@@ -192,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)
@@ -219,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 20ffc59..040860c 100644
--- a/tests/unit/SimpleOperators.swift
+++ b/tests/unit/SimpleOperators.swift
@@ -19,22 +19,41 @@ import IndefiniteObservable
 
 // Simple operators used by the tests.
 
-extension IndefiniteObservable {
-  public func map<U>(_ transform: @escaping (T) -> U) -> IndefiniteObservable<U> {
-    return IndefiniteObservable<U> { observer in
-      return self.subscribe {
+public final class ValueObserver<T>: Observer {
+  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) -> ValueObservable<U> {
+    return ValueObservable<U> { observer in
+      return self.subscribe(observer: ValueObserver<T> {
         observer.next(transform($0))
-      }.unsubscribe
+      }).unsubscribe
     }
   }
 
-  public func filter(_ isIncluded: @escaping (T) -> Bool) -> IndefiniteObservable<T> {
-    return IndefiniteObservable<T> { observer in
-      return self.subscribe {
-        if isIncluded($0) {
+  // Only emit values downstream for which passesTest returns true
+  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
     }
   }
 }